Compare commits
42 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f9f4d263af | |||
| 0049b6ba84 | |||
| 4ec82c8f8f | |||
| bdc9b2aed5 | |||
| 621ceba331 | |||
| daf15e21d3 | |||
| 0133ce99d6 | |||
| 33fd7bf2e1 | |||
| ec6076700b | |||
| 3d279cd7d8 | |||
| 6b66febba5 | |||
| c8531d8a14 | |||
| 9726e31677 | |||
| 4182070e76 | |||
| 3c974009be | |||
| 298d7c4564 | |||
| c7591bca0f | |||
| 160137ee2d | |||
| e044c63546 | |||
| c3aeee6323 | |||
| 88c2819edd | |||
| 866aeaf068 | |||
| c05b979b01 | |||
| 3f81734e69 | |||
| adcbbb66a2 | |||
| a13c535919 | |||
| 27e336eee9 | |||
| 71d3cf0590 | |||
| ee140cfd40 | |||
| 1cf1d228bc | |||
| 9cc7b5b050 | |||
| 3f3dc3f7ad | |||
| 5fad2fc575 | |||
| 1b81d1a30a | |||
| 6cf74672a5 | |||
| 5a925457b0 | |||
| 8e82742900 | |||
| f327861a00 | |||
| fd3fdb6cf7 | |||
| 346208a272 | |||
| c7cc4d78f7 | |||
| bf8178354f |
@@ -49,12 +49,20 @@ 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','maintain','event-identity','--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)
|
||||
subprocess.run([python,'-I','-B','-m','paper_atlas','audit','public-openreview','--help'],cwd=outside,check=True)
|
||||
subprocess.run([python,'-I','-B','-m','paper_atlas','coverage','--help'],cwd=outside,check=True)
|
||||
subprocess.run([python,'-I','-B','-m','paper_atlas','audit','histories','--help'],cwd=outside,check=True)
|
||||
subprocess.run([python,'-I','-B','-m','paper_atlas','curate','identifiers','--help'],cwd=outside,check=True)
|
||||
subprocess.run([python,'-I','-B','-c','from paper_atlas.postgres.writer import PostgresStore, WriteConnection'],cwd=outside,check=True)
|
||||
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
|
||||
|
||||
@@ -51,7 +51,8 @@ PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service fo
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --paper 123
|
||||
|
||||
# 列出全部目标会议,包括尚未入库的项目
|
||||
paper-atlas coverage --db /data/paper-index/papers.sqlite \
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas coverage \
|
||||
--pg-service foundry-research --pg-schema research \
|
||||
--out /data/paper-index/collected/venue-coverage-current.json
|
||||
|
||||
# 采集与导入分开;这里只采集到独立文件
|
||||
@@ -86,3 +87,9 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \
|
||||
CI 在已授权的远程 runner 上运行测试、构建 wheel,并从源码目录外验证安装后的命令及随包资源。数据库全量去重和覆盖率是独立的数据验收步骤,不因代码测试通过而自动完成。
|
||||
|
||||
开发通过工作分支和 PR 交付,由用户手动合并。
|
||||
|
||||
历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划,再按计划哈希执行;保留论文与投稿决定来源,不重导入。
|
||||
同年多届的 `event_key`、显式迁移、旧文件刷新与查询边界见 [会议事件身份](docs/event-identity.md)。
|
||||
研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回;两者独立查询,保留证据及完整更正历史,不从会议轨道推断研究类型。
|
||||
|
||||
整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.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.
|
||||
@@ -95,6 +95,12 @@ paper-atlas audit public-openreview --db /data/paper-index/papers.sqlite \
|
||||
|
||||
输出按会议、届次及 main、position、datasets_benchmarks 分文件。workshop 默认进入单独的 held 清单;有明确 workshop 名称时可通过 `--include-workshops` 单列采集,不合并到主会。Retraction、未映射的轨道或负面标签也保留在 `held-records.jsonl`,不强改为 Reject 或 Withdrawn。当前决定表没有独立 retracted 状态,因此撤回发表记录仍需后续专门建模。
|
||||
|
||||
`parser_version=2` 还会将明确负面标签但标题缺失、空白或 null 的记录放入 `held-records.jsonl`,保留原始状态、forum ID、固定来源版本和行号;不制造标题或导入不可识别的论文,也不让这一条缺失阻断其他可用记录。held 文件哈希与各原因数量保存在 inventory;非文本标题、错误身份、冲突届次和哈希不符仍使整批失败。旧年 ICLR 镜像的明确 `Withdraw` 标签归为 `withdrawn`,`Workshop`、`Poster Workshop`、`Oral Workshop` 均不作为拒稿。采集文件与导入主库仍是两个独立步骤。
|
||||
|
||||
`parser_version=3` 增加五组已核实身份的 ICLR workshop 标签:2024 ME-FoMo / GenAI4DM 的 Withdrawn Submission、2025 AgenticAI 的 Reject / LMRL 的 Withdrawn Submission,以及 2026 LLM Reasoning 的 Withdrawn Submission。仅在 `--include-workshops` 下输出,并分别保存 `workshop_me_fomo`、`workshop_genai4dm`、`workshop_agenticai`、`workshop_lmrl`、`workshop_llm_reasoning`;默认仍暂存。标签必须与原文件会议、年份和原始 workshop track 一致,错年/错会议使采集失败,错轨道单独暂存,未知标签继续待核查,不用宽泛的 Reject/Withdraw 正则猜测。
|
||||
|
||||
[OpenReview 官方 ICLR 活动目录](https://openreview.net/venue?id=ICLR.cc) 确认上述独立 workshop 身份;[ME-FoMo 2024](https://sites.google.com/view/me-fomo2024/call-for-papers)、[GenAI4DM 2024](https://sites.google.com/view/genai4dm-iclr2024)、[AgenticAI 2025](https://iclragenticai.github.io/) 和 [LLM Reasoning 2026](https://sites.google.com/view/iclr-2026-llmreasoning) 的活动网站也提供对应投稿入口。输出在 metadata 保存 `workshop_venueid` 和身份目录证据,决定仍只标为固定镜像的公开分类断言,保留原始标签、forum、来源行号与哈希,不声称核验原始决定签名。2026-11-08 核查时部分原站记录页要求浏览器验证;身份目录和主办方入口可读,不代表逐篇原始决定已可访问,也不外推其他 workshop 的公开政策。采集后仍须经过 PostgreSQL 身份/决定冲突预检、串行导入和独立验收。
|
||||
|
||||
该来源明确标记为 `curated_public_status_assertion`:保留源 commit、文件哈希、行号、原标签和 forum ID,不声称已核验原始 PC 签名。只保存标题、摘要、公开作者名、关键词、领域、索引与决定来源,不缓存镜像中的个人资料、邮箱、性别、评审内容或 PDF。所有输出清单均为 partial;盘点中的零条记录仅描述这个固定镜像。
|
||||
|
||||
入库前审计核验文件哈希、数量、轨道和来源声明,并用真实 forum ID 对比同届同轨道已有投稿。存在不同决定时阻止整批导入,交由来源和时序核查;另一会议的后续录用属于独立投稿历史,不构成该次拒稿的冲突。同一 forum 的镜像不新增投稿事件,但论文实体的跨平台去重仍须在全量结束后核验。审计通过的报告包含带 SHA-256 的导入队列;应在单写入维护锁下及时串行导入,主库变化后重新审计。
|
||||
@@ -180,6 +186,10 @@ paper-atlas collect dblp \
|
||||
|
||||
输出按会议/年份/track 分文件,附 manifest、`dblp-normalization.json` 范围审查统计和带哈希的 `dblp-import-queue.json`。不能识别的书目、年份或归属会逐类登记,不默认算作主会。DBLP 不含摘要,因此输出摘要为空、清单为 partial;需要通过有证据的跨来源关联补齐论文实体摘要。书目发表证据与原始审稿决定分别保留,不依据书目缺席推断拒稿。已存在其他出版来源的届次,还须先回填其权威身份并核验镜像记录,不能直接累加投稿数。
|
||||
|
||||
早期 AISTATS 1995/1997/1999/2001/2003/2005 和 COLT 1988–1991 的主系列名称本身含 Workshop,不能当作附属 workshop 漏掉。识别要求已核对的完整 proceedings key、会议书目名、原始标题前缀、届数和事件年一致;轨道名为 `main` 表示主系列,保留原始 Workshop 名称及官方目录证据,不改造其历史名称或出版形式。参见 [AISTATS 官方历史目录](https://aistats.org/other.html)、[COLT 协会目录](https://www.learningtheory.org/)。1995 AISTATS 的会议年与1996图书出版年分开;[PMLR R0](https://proceedings.mlr.press/r0/) 是后来重发的 pre-proceedings,不与 DBLP 图书条目盲目逐篇等同。
|
||||
|
||||
真正附属的 AISTATS 2010 Active Learning and Experimental Design workshop 仍单列。COLT 2003 与 Kernel Workshop 的 [混合论文集](https://doi.org/10.1007/b12006) 缺少逐篇章节信息,使用 `proceedings_unspecified`,不能整卷改成 main 或附属 workshop。输出保存 `track_scope_note` 与 `track_scope_evidence_url`,导入后仍可追查分类理由;这些出版证据不是真实拒稿或原始决定签名,也不自动修正既有主库记录。
|
||||
|
||||
部分早期 SIGIR、CIKM、IJCAI 出版卷使用两位年份键;只有论文集标题明确给出唯一完整年份且与短键相符时才采用,不推测世纪。WWW 特殊兴趣、Alternate track 单列;同卷混合 Applied Data Science、Demo、Industrial 且不能逐篇区分时使用 `proceedings_unspecified`,不把整卷归为某一个具体 track。
|
||||
|
||||
书目中的前后页、目录组合和明确会务标题不产生论文录用记录;缺作者的赞助、委员会、活动汇总等标题单独排除。WWW 2019 companion 的匿名合成活动条目归入待核查材料,原始快照仍保留。范围报告分别统计这些原因;单纯缺作者并不排除真实研究论文,也不补造作者。
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
# 同年多届会议的事件身份
|
||||
|
||||
`venue`、`year`、`track` 描述系列、年份和投稿轨道;`event_key` 区分同系列、同年、同轨道的实际会议事件。例如 WWW 1995 的春季与秋季会议分别使用 `www-1995-spring` 和 `www-1995-fall`,两者仍是 `main`。相同论文可以在两届有不同投稿与决定;论文去重不会把这两次投稿合成一次。
|
||||
|
||||
普通单届数据继续使用空键。新输入在同一组已有具名事件或已知同年多届时必须提供明确键;不能任选第一个 edition。已有来源刷新可依据自己的原投稿关系回到原事件,不能被显式改到另一事件。精确固定 DBLP 快照、WWW Fall 书目名、1995 年及 `www1995f.html` 卷共同支持旧秋季文件的归属;不按标题猜年份或季节。
|
||||
|
||||
规范 JSONL 行及 manifest 都可提供 `event_key`,非空值限制为最多80个小写字母、数字及 `._:-`,首字符为字母或数字。manifest 提供键时,每行必须一致;一份文件仍仅对应一个事件及轨道。跨行冲突会回滚整份业务导入。身份回填同样按事件选择,不能借机把来源移到另一届。
|
||||
|
||||
## 新样本与既有数据
|
||||
|
||||
新建 SQLite 样本的 `editions` 唯一键为 `(venue_id, year, track, event_key)`。已有 SQLite/PostgreSQL 不会因普通读写命令自动升级。其普通单届导入继续可用;WWW1994/1995或带事件键的导入会要求先迁移。冻结的 SQLite 生产备份仍受 cutover 标记保护,不能写入或升级。
|
||||
|
||||
先在 PostgreSQL 生成只读计划:
|
||||
|
||||
```sh
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain event-identity \
|
||||
--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`;缺证或混合卷会拒绝计划。本文命令不表示生产已经迁移。
|
||||
|
||||
应用时传入同一文件及命令输出的文件 SHA256:
|
||||
|
||||
```sh
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain event-identity \
|
||||
--pg-service foundry-research --pg-schema research \
|
||||
--apply-plan /data/paper-index/event-plan.json --plan-sha256 PLAN_FILE_SHA256 \
|
||||
--out /data/paper-index/event-applied.json
|
||||
```
|
||||
|
||||
原生 PostgreSQL writer 使用现有 schema advisory lock,并在事务中锁定 editions,核对计划后添加列及替换唯一约束。SQLite 样本使用独占写事务,重建父表时保留 IDs,并在提交前检查外键;不级联删除子记录。额外列、未知唯一约束或 SQLite 额外索引会拒绝自动升级,须单独评估。未改变原版四列的排序内容哈希、任何业务表数量或目标投稿/决定/来源内容,才可提交;失败回滚。迁移标记与 DDL 同事务。原计划立即重试为 `already_applied`;若后续业务已经变化,应重新生成计划,不重放旧计划。
|
||||
|
||||
ECML 与 PKDD 的联合举办前事件已有不同 venue/edition,保持原身份即可。这项迁移不会把它们合并,也不会创建新的论文、投稿或决定。WWW 春季及1994两届的逐篇采集、来源核验和导入仍须另做。
|
||||
|
||||
## 查询与验收
|
||||
|
||||
```sh
|
||||
paper-atlas query --pg-service foundry-research --pg-schema research \
|
||||
--venue WWW --year 1995 --track main --event-key www-1995-fall
|
||||
```
|
||||
|
||||
详细投稿历史及 `known_acceptances` 返回 `event_key`。`--event-key` 可与 `--status` 一起限定同一事件中的最新已知决定。未迁移的旧库仍可普通查询,其事件键为空;不能声称已经区分季节。`audit histories` 的新库事件输出也保留键。
|
||||
|
||||
远程测试应覆盖同一作品的两届独立投稿、拒稿后另一届录用、原 IDs 与 Unicode/NUL 来源内容保留、旧文件刷新、歧义及错误键拒绝、整文件回滚、明确事件的身份回填、旧计划失效、重试与真实 PostgreSQL 迁移。生产应用后还须独立核验约束、来源和历史、全文索引及 foundry-core 健康;单元测试不替代生产或最终全库去重验收。
|
||||
@@ -0,0 +1,67 @@
|
||||
# Repairing joint conference membership
|
||||
|
||||
EURAI's [official history](https://www.eurai.org/ecai/history) identifies ECAI 2018
|
||||
and 2022 as joint IJCAI/ECAI conferences. The pinned DBLP volumes
|
||||
`conf/ijcai/2018` and `conf/ijcai/2022` use `IJCAI` as their booktitle and title.
|
||||
Existing records can therefore already be present under IJCAI while an ECAI query
|
||||
shows a missing year. Adding ECAI as a global alias to the ordinary IJCAI venue
|
||||
would incorrectly include all other IJCAI years.
|
||||
|
||||
The existing canonical `ECAI+IJCAI` venue has both member aliases. This maintenance
|
||||
command changes only the venue reference of the two existing `main` edition rows.
|
||||
Edition IDs, submission IDs, decisions, source text, authors, identifiers, tags,
|
||||
source links and coverage records stay intact. It does not merge or create papers
|
||||
or submissions, resolve per-paper tracks, or turn reviews into negative decisions.
|
||||
The DBLP collector applies the joint mapping only to these two exact main volume
|
||||
keys; original booktitle/title remain in source metadata with the membership evidence.
|
||||
The shared ingestion entry point also resolves these exact pinned source volumes before
|
||||
creating an edition, so refreshing an older normalized file cannot recreate a separate
|
||||
IJCAI edition. Other sources, years, workshop volumes and additional tracks retain their existing rules.
|
||||
|
||||
## Review a read-only plan
|
||||
|
||||
Use a runtime containing this repair command. The configured PostgreSQL writer lock
|
||||
is used only for the apply operation; do not replace a newer ingestion runtime with
|
||||
an older main-based feature branch.
|
||||
|
||||
```sh
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \
|
||||
--pg-service foundry-research --pg-schema research --out joint-plan.json
|
||||
```
|
||||
|
||||
The preflight requires the existing joint venue and exactly its two member aliases,
|
||||
one nonempty unambiguous edition per year, and matching pinned DBLP volume evidence
|
||||
for every existing submission. A colliding target edition, missing provenance or
|
||||
unexpected source scope fails closed. The plan binds current database/schema, IDs,
|
||||
counts and full fingerprints of affected papers, identifiers, source records, tags,
|
||||
submissions, decisions and their provenance, and coverage rows. Observation time is
|
||||
recorded separately from conference year. These are scoped checks, not a final global
|
||||
identity or conference completeness audit.
|
||||
|
||||
## Apply the exact reviewed plan
|
||||
|
||||
Compute the file's SHA256 after review and pass it explicitly:
|
||||
|
||||
```sh
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \
|
||||
--pg-service foundry-research --pg-schema research --plan joint-plan.json \
|
||||
--plan-sha256 "$PLAN_SHA256" --out joint-applied.json
|
||||
```
|
||||
|
||||
The plan hash is checked before opening a writer. Current evidence is checked again
|
||||
under the dataset writer lock; both edition updates and an audit marker commit in one
|
||||
transaction. Protected row hashes are checked again before commit. Changed evidence
|
||||
requires a fresh reviewed plan. Existing output reports are preserved; choose a new
|
||||
path for another run. SQLite samples use `--db sample.sqlite`; the frozen production
|
||||
backup remains protected.
|
||||
|
||||
If a response is lost after commit, rerun with the same plan and a new output path.
|
||||
When protected evidence is unchanged and both editions already point to the joint
|
||||
venue, the result is `already_joint` with zero updates. It does not replay imports or
|
||||
regenerate source records. Keep the plan, apply report and independent postcheck
|
||||
alongside collection evidence. The audit marker is `joint_ijcai_ecai_2018_2022_v1`.
|
||||
|
||||
Publication counts can differ between the DBLP volume and the current publisher
|
||||
page. This membership repair preserves all existing rows rather than deleting entries
|
||||
to force an aggregate match; section-level main/invited/other classification and
|
||||
publication completeness require separate per-paper evidence.
|
||||
+38
-2
@@ -97,13 +97,16 @@ paper-atlas audit arxiv-tags --db /data/paper-index/papers.sqlite \
|
||||
采集文件、已入库的来源记录和去重论文实体是不同的计数。来源可能覆盖同一篇论文,同一篇论文也可能有多次投稿;不能把各批文件数量相加作为独立论文总数。实时统计与覆盖报告使用:
|
||||
|
||||
```sh
|
||||
paper-atlas query --db /data/paper-index/papers.sqlite --stats
|
||||
paper-atlas coverage --db /data/paper-index/papers.sqlite \
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --pg-schema research --stats
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas coverage \
|
||||
--pg-service foundry-research --pg-schema research \
|
||||
--out /data/paper-index/collected/venue-coverage-current.json
|
||||
```
|
||||
|
||||
manifest 和采集器状态清单用于确认文件是否齐全及是否缺摘要;`ingest_runs` 和协调器日志用于确认主库导入是否成功。官方程序、书目目录和公开投稿快照分别保留,未出现在某份清单不等于被拒稿;第三方快照也不能代表平台当前全量状态。代码测试成功不能代替最终全库去重、完整性和逐届覆盖核验。
|
||||
|
||||
`coverage` 的所有计数来自同一个只读快照。`latest_observed_status_counts` 按每次投稿的最新观察决定统计,`decision_observation_counts` 保留所有历史观察;重复来源不增加投稿数,缺少决定的投稿计入最新状态 unknown,但不凭空制造 unknown 决定。`source_record_counts` 统计来源在该届次的实际记录与缺摘要数,和 `sources` 中逐清单的预期/导入数量分开;身份补充清单与原始论文集清单可以描述同一条记录。来源缺摘要与论文实体从其他来源获得摘要也分别报告。报告仍将历史覆盖标为待独立审计,不能用某个清单 complete 或没有拒稿记录推断全会议覆盖。
|
||||
|
||||
当前代码与合并状态以 [Pull Requests](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls) 为准;CI 结果以对应提交的 Actions 检查为准。工作分支交付后等待用户手动合并。合并代码不自动部署、不自动启动采集,也不代表数据批次已完成。
|
||||
## 修正已有 PMLR 投稿的轨道
|
||||
|
||||
@@ -120,3 +123,36 @@ paper-atlas maintain pmlr-tracks --db /data/paper-index/papers.sqlite \
|
||||
检查报告后,在同一命令添加 `--apply` 并使用新的 `--out` 路径执行迁移。计划使用共享维护锁,实际迁移使用独占锁;已有 arXiv 类别修复或其他写入运行时会拒绝启动,不会停止或重启它们。一次事务内核对整个来源集合、原始题目/摘要/作者、forum 身份、已有事件及目标冲突,更新原投稿的 edition 和身份归属,回填官方轨道出处及逐轨道覆盖数,最后复查轨道与外键后提交。论文、来源、投稿、决定的 ID 及决定观察均保留,不重新创建事件、不做实体合并、不改论文正文元数据或标签。
|
||||
|
||||
报告记录每个迁移事件的旧 edition、原身份、完整决定观察及新轨道证据。`planned` 是未执行的计划;`complete` 才表示事务已提交;`error` 表示执行失败。若进程在提交前后中断而报告停在 `prepared`,先对照主库和报告核查,不根据日志缺失自行重启任务。新路径重跑会重新核对当前数据库,已正确的轨道不会再次迁移。共享多个来源、重复目标事件或冲突决定会停止迁移,需另行核对,不能直接删掉其中一条。
|
||||
|
||||
## 追加同一目录的部分记录
|
||||
|
||||
默认 `import records` 把 manifest 当成一份来源快照,覆盖统计使用该文件的数量。
|
||||
若目录已有2条,之后仅补入1条,不能把这一小批当成完整快照,否则会把目录计数改成1。
|
||||
为已核验的部分补充批次,在 manifest 显式声明:
|
||||
|
||||
```json
|
||||
{
|
||||
"coverage_mode": "incremental",
|
||||
"coverage_previous_count": 2,
|
||||
"coverage_previous_expected_count": 2,
|
||||
"coverage_expected_count": 4,
|
||||
"status": "partial",
|
||||
"count": 1
|
||||
}
|
||||
```
|
||||
|
||||
这只是新增字段示例,manifest仍须保留source、venue、year、track、list_url、文件哈希等来源证据。
|
||||
`count`是本文件行数,`coverage_previous_count`是导入前该来源/届次的实际记录数,
|
||||
`coverage_previous_expected_count`是旧覆盖行的expected_count(未知时用null),
|
||||
`coverage_expected_count`是独立目录证据支持的整份来源预期数,不能小于已知旧预期数。
|
||||
这个例子成功后覆盖统计为预期4、已导入3;文件checkpoint的records_written仍为1。
|
||||
刷新相同记录不重复累加。初次导入部分来源可声明previous_count为0、previous_expected_count为null。
|
||||
|
||||
只支持当前最多一个规范目录URL且旧覆盖状态为partial的来源/届次;多目录、旧状态complete、
|
||||
旧计数与实际来源记录不符,或前置计数已变化,都会中止,先核对来源范围或重新取得完整快照。
|
||||
该模式在写入前检查父状态,并在同一文件事务中按实际来源记录重新计数、检查总量界限和乐观更新覆盖行。
|
||||
失败会回滚本文件的论文、来源、决定、标签和覆盖变化,保留失败运行记录;更早已提交文件不重放。
|
||||
增量模式始终保留partial,不能靠一个小批次推断完整论文集、摘要或公开拒稿覆盖。
|
||||
|
||||
仍需先独立核对新增记录身份与决定,按原生单写入保护串行导入,再独立核验数据与原有历史。
|
||||
这些计数检查不证明外部目录完整,也不代替最终全库去重。
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
# 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.
|
||||
|
||||
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.
|
||||
+76
-6
@@ -2,6 +2,55 @@
|
||||
|
||||
先在已确认的远程环境安装项目,使用 `paper-atlas`。从源码运行时,将命令前缀替换为 `python3 -B -m paper_atlas`,其余参数相同。数据目录位于仓库外。
|
||||
|
||||
## 按证据补充稳定标识
|
||||
|
||||
`paper-atlas curate identifiers --plan /data/review/plan.json --output /data/review/prepared`
|
||||
只生成新的 JSONL、manifest、证据副本及 `curation-report.json`,不会连接数据库。
|
||||
输出目录必须不存在,父目录必须已存在;任何输入或断言失败,都不发布本批输出。
|
||||
原始文件不变,未选中的记录逐字节保留。题名、作者、年份、页码、投稿决定、标签和既有 metadata
|
||||
均不重写,仅补充 `identifiers` 并追加 `metadata.identifier_curation` 证据链。
|
||||
|
||||
计划格式如下(哈希占位符须替换为真实 SHA256,所有路径按命令当前目录解析):
|
||||
|
||||
```json
|
||||
{
|
||||
"format_version": 1,
|
||||
"inputs": [{
|
||||
"file": "/data/review/uai2001.jsonl",
|
||||
"sha256": "<JSONL 文件 SHA256>",
|
||||
"manifest_sha256": "<同名 .manifest.json 文件 SHA256>"
|
||||
}],
|
||||
"assertions": [{
|
||||
"source": "dblp",
|
||||
"external_id": "conf/uai/example",
|
||||
"record_sha256": "<原始物理行字节 SHA256,包含该行原有换行符>",
|
||||
"identifiers": {"arxiv": "1408.1485"},
|
||||
"evidence": [{
|
||||
"file": "/data/review/primary-metadata.json",
|
||||
"sha256": "<证据 JSON 文件 SHA256>",
|
||||
"url": "https://example.org/primary-record",
|
||||
"checked_at": "2026-11-08T08:00:00Z",
|
||||
"note": "说明人工核对的题名、作者、出版版本和稳定标识证据"
|
||||
}]
|
||||
}]
|
||||
}
|
||||
```
|
||||
|
||||
输入 manifest 必须包含 `source/venue/year/track/count/list_url/file_sha256`,且与逐条记录一致。
|
||||
断言必须唯一、准确命中一条记录。仅支持无版本后缀的 arXiv 编号与小写规范 DOI;
|
||||
已有同类标识不允许覆盖,相同值可以补记证据。证据只接受不超过 16 MiB 的 JSON 元数据文件,
|
||||
来源 URL 要求 HTTPS 且不含凭据,核查时间必须带时区;不下载论文 PDF。
|
||||
|
||||
`verification_scope=input_and_evidence_integrity_only` 和 `semantic_identity_verified=false`
|
||||
明确表示哈希校验不能证明两条记录属于同一作品。请先独立核对版本关系;仅年份、页码或相同题名
|
||||
不能作为合并依据。存在来源冲突的行应继续暂缓。新 manifest 保留原来的 partial 状态及摘要缺失数,
|
||||
不会把补充编号伪装成来源覆盖或摘要补全。证据副本按 SHA256 保存在输出目录 `evidence/`,
|
||||
原计划也保存在 `plan.json`,供之后复查。
|
||||
|
||||
文件生成后仍需另做当前 PostgreSQL 身份、投稿和决定冲突预检;导入运行时必须包含
|
||||
“稳定标识优先、多个标题候选则整文件回滚”的身份保护,再以 `import records` 独立导入并验收。
|
||||
该命令不会自动导入或合并实体,也不代表最终全库去重通过。
|
||||
|
||||
## PostgreSQL 查询
|
||||
|
||||
在远程环境安装 `.[postgres]`,通过权限为 `0600` 的 libpq service 文件连接,不把密码放入 URL、命令行或仓库。
|
||||
@@ -79,22 +128,43 @@ paper-atlas maintain merge --db /data/paper-index/samples/example.sqlite \
|
||||
## 4. 查询与分析前检查
|
||||
|
||||
```sh
|
||||
paper-atlas query --db /data/paper-index/papers.sqlite --query '"language model"' --limit 20
|
||||
paper-atlas query --db /data/paper-index/papers.sqlite --tag arxiv:category=cs.AI
|
||||
paper-atlas query --db /data/paper-index/papers.sqlite --venue ICLR --year 2025 --status rejected
|
||||
paper-atlas query --db /data/paper-index/papers.sqlite --paper 123
|
||||
paper-atlas coverage --db /data/paper-index/papers.sqlite --out /data/paper-index/collected/coverage.json
|
||||
paper-atlas query --pg-service foundry-research --query '"language model"' --limit 20
|
||||
paper-atlas query --pg-service foundry-research --tag arxiv:category=cs.AI
|
||||
paper-atlas query --pg-service foundry-research --venue ICLR --year 2025 --status rejected
|
||||
paper-atlas query --pg-service foundry-research --paper 123
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas coverage \
|
||||
--pg-service foundry-research --pg-schema research --out /data/paper-index/collected/coverage.json
|
||||
```
|
||||
|
||||
`--status rejected` 筛选某次投稿的最新已知决定。它并不表示论文在所有会议均未录用;分析转投情况时查看该论文的完整投稿历史。年份表示会议届次,观察时间和决定日期分别保留。
|
||||
|
||||
覆盖报告使用只读一致性快照,标明实际数据库和 schema,并列出零入库目标与注册表外的已入库会议。逐届、逐 track 分别列出最新观察状态的投稿数、全部决定观察数、各来源记录数,以及来源摘要和论文实体摘要的缺失数。摘要缺失指 NULL 或空字符串;有文字不代表内容已经完整核验。同一来源的多个清单 URL 可以重叠,不能把清单计数相加当作来源总量;零条拒稿也不证明该会议没有公开拒稿。SQLite 样本仍可显式使用 `--db`,生产报告应查询当前 PostgreSQL。
|
||||
|
||||
例如,查找在 ICLR 有拒稿记录、但在其他已知投稿中已有录用的论文:
|
||||
|
||||
```sh
|
||||
paper-atlas query --db /data/paper-index/papers.sqlite \\
|
||||
paper-atlas query --pg-service foundry-research \
|
||||
--venue ICLR --status rejected --outcome accepted
|
||||
```
|
||||
|
||||
`known_acceptances` 列出目前仍为录用状态的各次投稿及会议、年份、track 和决定证据,`publication_outcome` 汇总为 accepted 或 no_known_acceptance。汇总会检查所有会议和 track,不把某次拒稿覆盖到整篇论文;同一次投稿的后续撤稿也会更新该次状态并保留原决定。no_known_acceptance 只表示现有来源中没有查到录用,不能称为“从未录用”。
|
||||
|
||||
`--stats` 中 papers 是当前规则下的论文实体数;source_records 是来源记录数;decisions 是决定观察数。一个投稿可能有多个来源和多条观察,不能相加当作论文数。
|
||||
|
||||
已有稳定标识或原来源记录匹配时,该身份优先;同标题、同作者的另一 arXiv 存档或扩展稿不会仅因相似标题被并入。没有身份锚点时,作者佐证的精确标题只有一个候选才可匹配;多个候选会报 `Ambiguous paper identity`,整个采集文件回滚并记录失败运行。应补有证据的标识或先核查版本关系,不删除重复候选或任取第一个。一个新来源明确同时携带两个既有稳定标识时,仍可按既有规则归并并保留投稿、标签和合并日志;未归并的同标题关系继续保留为待核查候选。这些规则不证明全库去重完成。
|
||||
|
||||
## 5. 投稿历史关系审计
|
||||
|
||||
```sh
|
||||
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas audit histories \
|
||||
--pg-service foundry-research --pg-schema research --examples 20 \
|
||||
--out /data/paper-index/collected/history-audit.json
|
||||
```
|
||||
|
||||
使用一个只读一致性快照,区分论文实体、投稿和决定观察;分别统计同次投稿曾出现负面与录用观察、不同投稿共存负面与当前已知录用、不同会议之间的共存。workshop/main 是不同 track,属于同一会议时不算跨会议;会议别名也不会增加一个会议。当前已知录用使用与 `query --outcome` 相同的“每次投稿最新观察”规则,后续撤稿的该次投稿不再算当前录用,但旧录用决定仍保留。
|
||||
|
||||
报告不凭观察时间、届次年份或行顺序断言“拒稿发生在录用之前”。样例保留真实 `decision_date`、`observed_at`、原始标签、证据 URL 和决定来源,供进一步核查;日期缺失仍保持未知。`negative` 统计包含 rejected、desk_rejected 和 withdrawn,其原始状态不互相替换。
|
||||
|
||||
总数覆盖快照内全部投稿,样例由 `--examples` 限定为 0–100 篇;每篇最多展示 20 次投稿、每次最多 50 条决定、每条决定最多 100 个来源,并明确标记截断。样例优先保留负面/录用的不同投稿对。四项来源关系检查报告缺失关联和来源论文归属不一致,不自动修复。
|
||||
|
||||
这些统计依赖现有论文身份关联,不证明实体合并正确,也不证明原始决定签名、公开拒稿全集、标签类型或最终全库去重已完成。SQLite 独立样本可使用 `--db`;冻结备份不能代表当前 PostgreSQL 的历史情况。
|
||||
|
||||
+120
-27
@@ -1,45 +1,138 @@
|
||||
"""Show every target venue, including those with no imported records."""
|
||||
"""Describe imported venue coverage without implying a complete census."""
|
||||
import argparse
|
||||
from paper_atlas.config import DEFAULT_REGISTRY
|
||||
from collections import defaultdict
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
|
||||
from paper_atlas.config import DEFAULT_REGISTRY
|
||||
from paper_atlas.db.backend import database_arguments, open_read_connection
|
||||
from paper_atlas.db.submissions import timestamp
|
||||
from paper_atlas.querying import LATEST_DECISION
|
||||
|
||||
|
||||
def report(db,registry):
|
||||
registry=json.loads(Path(registry).read_text());targets=registry['venues']
|
||||
names=[row['acronym'].casefold() for row in targets]
|
||||
if len(names)!=len(set(names)):raise ValueError('Duplicate target venue')
|
||||
scope='(v.acronym=? COLLATE NOCASE OR EXISTS(SELECT 1 FROM venue_aliases a WHERE a.venue_id=v.id AND a.alias=? COLLATE NOCASE))'
|
||||
c=sqlite3.connect(Path(db).resolve().as_uri()+'?mode=ro',uri=True);c.row_factory=sqlite3.Row
|
||||
STATUSES = ('accepted', 'rejected', 'desk_rejected', 'withdrawn', 'under_review', 'unknown')
|
||||
|
||||
|
||||
def _status_counts(connection, *, latest):
|
||||
if latest:
|
||||
query = '''SELECT s.edition_id,COALESCE(d.status,'unknown') AS status,count(*) AS count
|
||||
FROM submissions s LEFT JOIN decisions d ON d.submission_id=s.id AND ''' + LATEST_DECISION
|
||||
else:
|
||||
query = '''SELECT s.edition_id,COALESCE(d.status,'unknown') AS status,count(*) AS count
|
||||
FROM submissions s JOIN decisions d ON d.submission_id=s.id'''
|
||||
counts = defaultdict(dict)
|
||||
for row in connection.execute(query + " GROUP BY s.edition_id,COALESCE(d.status,'unknown')"):
|
||||
counts[row[0]][row[1]] = row[2]
|
||||
return counts
|
||||
|
||||
|
||||
def report(db, registry, *, pg_service=None, pg_schema=None):
|
||||
started_at = timestamp()
|
||||
registry = json.loads(Path(registry).read_text())
|
||||
targets = registry['venues']
|
||||
names = [row['acronym'].casefold() for row in targets]
|
||||
if len(names) != len(set(names)):
|
||||
raise ValueError('Duplicate target venue')
|
||||
scope = '''(lower(v.acronym)=lower(?) OR EXISTS(
|
||||
SELECT 1 FROM venue_aliases a WHERE a.venue_id=v.id AND lower(a.alias)=lower(?)))'''
|
||||
connection = open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)
|
||||
if not pg_service:
|
||||
connection.row_factory = sqlite3.Row
|
||||
try:
|
||||
c.execute('BEGIN');venues=[]
|
||||
if pg_service:
|
||||
identity = connection.execute('SELECT current_database(),current_schema()').fetchone()
|
||||
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
|
||||
# for each registry target caused repeated scans on the production DB.
|
||||
edition_counts = {}
|
||||
for row in connection.execute('''SELECT s.edition_id,count(s.id) AS submissions,
|
||||
count(DISTINCT s.paper_id) AS papers,
|
||||
count(DISTINCT CASE WHEN p.abstract IS NULL OR p.abstract='' THEN s.paper_id END)
|
||||
AS papers_without_abstract_text
|
||||
FROM submissions s LEFT JOIN papers p ON p.id=s.paper_id GROUP BY s.edition_id'''):
|
||||
entry = dict(row)
|
||||
edition_counts[entry.pop('edition_id')] = entry
|
||||
source_counts = defaultdict(list)
|
||||
for row in connection.execute('''SELECT s.edition_id,so.name AS source,
|
||||
count(DISTINCT sr.id) AS source_records,
|
||||
count(DISTINCT CASE WHEN sr.abstract IS NULL OR sr.abstract='' THEN sr.id END)
|
||||
AS source_records_without_abstract_text
|
||||
FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id
|
||||
JOIN source_records sr ON sr.id=ss.source_record_id JOIN sources so ON so.id=sr.source_id
|
||||
GROUP BY s.edition_id,so.name ORDER BY s.edition_id,so.name'''):
|
||||
entry = dict(row)
|
||||
source_counts[entry.pop('edition_id')].append(entry)
|
||||
venues, covered_venue_ids = [], set()
|
||||
for target in targets:
|
||||
acronym=target['acronym'];params=(acronym,acronym)
|
||||
editions=[]
|
||||
for row in c.execute('SELECT e.id,v.acronym AS canonical_venue,e.year,e.track,count(s.id) AS submissions,count(DISTINCT s.paper_id) AS papers FROM editions e JOIN venues v ON v.id=e.venue_id LEFT JOIN submissions s ON s.edition_id=e.id WHERE '+scope+' GROUP BY e.id ORDER BY e.year,e.track',params):
|
||||
edition=dict(row)
|
||||
edition['sources']=[dict(source) for source in c.execute('SELECT so.name,es.status,es.expected_count,es.imported_count,es.list_url,es.last_success_at,es.last_error FROM edition_sources es JOIN sources so ON so.id=es.source_id WHERE es.edition_id=? ORDER BY so.name',(row['id'],))]
|
||||
acronym = target['acronym']
|
||||
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,''' + event + ''' AS event_key FROM editions e JOIN venues v ON v.id=e.venue_id
|
||||
WHERE ''' + scope + '''
|
||||
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}))
|
||||
covered_venue_ids.add(edition['venue_id'])
|
||||
edition['latest_observed_status_counts'] = {status: latest[row['id']].get(status, 0) for status in STATUSES}
|
||||
edition['decision_observation_counts'] = {status: observations[row['id']].get(status, 0) for status in STATUSES}
|
||||
edition['source_record_counts'] = source_counts[row['id']]
|
||||
edition['sources'] = [dict(source) for source in connection.execute('''SELECT so.name,
|
||||
es.status,es.expected_count,es.imported_count,es.list_url,es.last_success_at,es.last_error
|
||||
FROM edition_sources es JOIN sources so ON so.id=es.source_id
|
||||
WHERE es.edition_id=? ORDER BY so.name,es.list_url''', (row['id'],))]
|
||||
editions.append(edition)
|
||||
totals=c.execute('SELECT count(s.id),count(DISTINCT s.paper_id) FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE '+scope,params).fetchone()
|
||||
venues.append({'acronym':acronym,'field':target['field'],'imported_submissions':totals[0],'distinct_papers':totals[1],
|
||||
'history_status':'partial_history_pending_independent_audit' if totals[0] else 'no_imported_records',
|
||||
'discovery_sources':target.get('discovery_sources',[]),'editions':editions})
|
||||
return {'checked_at':timestamp(),'target_venue_count':len(targets),'venues_with_imported_records':sum(bool(v['imported_submissions']) for v in venues),
|
||||
'scope_note':registry['scope_note'],'interpretation':'All registry targets are listed. Edition source complete means that source file passed import checks; it does not prove full historical or abstract coverage. Papers can appear in multiple venues/years, so do not sum venue paper counts as a global total.',
|
||||
'venues':venues}
|
||||
finally:c.close()
|
||||
totals = connection.execute('''SELECT count(s.id),count(DISTINCT s.paper_id)
|
||||
FROM submissions s JOIN editions e ON e.id=s.edition_id
|
||||
JOIN venues v ON v.id=e.venue_id WHERE ''' + scope, params).fetchone()
|
||||
venues.append({'acronym': acronym, 'field': target['field'],
|
||||
'imported_submissions': totals[0], 'distinct_papers': totals[1],
|
||||
'history_status': 'partial_history_pending_independent_audit' if totals[0] else 'no_imported_records',
|
||||
'discovery_sources': target.get('discovery_sources', []), 'editions': editions})
|
||||
outside_registry = [dict(row) for row in connection.execute('''SELECT v.id AS venue_id,
|
||||
v.acronym,count(s.id) AS submissions,count(DISTINCT s.paper_id) AS papers
|
||||
FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
|
||||
GROUP BY v.id,v.acronym ORDER BY v.acronym''') if row[0] not in covered_venue_ids]
|
||||
return {'report_started_at': started_at, 'checked_at': timestamp(),
|
||||
'engine': 'postgresql' if pg_service else 'sqlite', 'database': database, 'schema': schema,
|
||||
'pg_service': pg_service, 'target_venue_count': len(targets),
|
||||
'venues_with_imported_records': sum(bool(v['imported_submissions']) for v in venues),
|
||||
'scope_note': registry['scope_note'],
|
||||
'interpretation': 'All registry targets are listed, including zero imports. '
|
||||
'Latest status counts use the latest observed decision for each attempt; decision observations retain history. '
|
||||
'Source record totals are per source and edition, not per list URL; multiple list entries may describe overlapping records. '
|
||||
'Missing abstract text means NULL or empty string; nonempty text does not prove a complete abstract. '
|
||||
'An edition source marked complete is not a proof of complete conference history or public rejection coverage. '
|
||||
'Zero imported rejections does not establish that no rejections are public. '
|
||||
'Papers can appear in several venues/years; do not sum venue paper counts as a global total.',
|
||||
'venues': venues, 'imported_venues_outside_registry': outside_registry}
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
def main(argv=None, *, prog=None):
|
||||
parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--db',required=True)
|
||||
parser.add_argument('--registry',default=str(DEFAULT_REGISTRY));parser.add_argument('--out')
|
||||
args=parser.parse_args(argv);result=report(args.db,args.registry);text=json.dumps(result,ensure_ascii=False,indent=2)
|
||||
parser = argparse.ArgumentParser(prog=prog)
|
||||
database_arguments(parser)
|
||||
parser.add_argument('--registry', default=str(DEFAULT_REGISTRY))
|
||||
parser.add_argument('--out')
|
||||
args = parser.parse_args(argv)
|
||||
result = report(args.db, args.registry, pg_service=args.pg_service, pg_schema=args.pg_schema)
|
||||
text = json.dumps(result, ensure_ascii=False, indent=2)
|
||||
if args.out:
|
||||
path=Path(args.out);stage=path.with_suffix('.part');stage.write_text(text+'\n');stage.replace(path)
|
||||
else:print(text)
|
||||
path = Path(args.out)
|
||||
stage = path.with_suffix('.part')
|
||||
stage.write_text(text + '\n')
|
||||
stage.replace(path)
|
||||
else:
|
||||
print(text)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
"""Audit linked submission histories without inferring event chronology."""
|
||||
import argparse
|
||||
from collections import Counter
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
|
||||
from paper_atlas.db.backend import database_arguments, open_read_connection
|
||||
from paper_atlas.db.submissions import timestamp
|
||||
from paper_atlas.querying import LATEST_DECISION
|
||||
|
||||
|
||||
ATTEMPTS = '''SELECT s.paper_id,s.id AS submission_id,e.venue_id,v.acronym AS venue,e.year,e.track,
|
||||
count(d.id) AS decision_observations,
|
||||
MAX(CASE WHEN d.status IN ('rejected','desk_rejected','withdrawn') THEN 1 ELSE 0 END)
|
||||
AS historical_negative,
|
||||
MAX(CASE WHEN d.status='accepted' THEN 1 ELSE 0 END) AS historical_acceptance,
|
||||
COALESCE(MAX(CASE WHEN ''' + LATEST_DECISION + ''' THEN d.status END),'unknown')
|
||||
AS latest_observed_status
|
||||
FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
|
||||
LEFT JOIN decisions d ON d.submission_id=s.id
|
||||
GROUP BY s.paper_id,s.id,e.venue_id,v.acronym,e.year,e.track
|
||||
ORDER BY s.paper_id,e.venue_id,s.id'''
|
||||
|
||||
|
||||
def _paper_counts(attempts):
|
||||
negative = sum(row['historical_negative'] for row in attempts)
|
||||
accepted = sum(row['latest_observed_status'] == 'accepted' for row in attempts)
|
||||
overlap = sum(row['historical_negative'] and row['latest_observed_status'] == 'accepted'
|
||||
for row in attempts)
|
||||
venues = {}
|
||||
for row in attempts:
|
||||
counts = venues.setdefault(row['venue_id'], [0, 0])
|
||||
counts[0] += row['historical_negative']
|
||||
counts[1] += row['latest_observed_status'] == 'accepted'
|
||||
return {
|
||||
'papers_with_submissions': 1,
|
||||
'papers_with_multiple_submissions': int(len(attempts) > 1),
|
||||
'papers_with_historical_negative': int(negative > 0),
|
||||
'papers_with_current_known_acceptance': int(accepted > 0),
|
||||
'papers_with_negative_and_acceptance_in_distinct_attempts': int(negative * accepted > overlap),
|
||||
'papers_with_negative_and_acceptance_in_distinct_venues':
|
||||
int(negative * accepted > sum(n * a for n, a in venues.values())),
|
||||
}
|
||||
|
||||
|
||||
def _example(connection, attempts):
|
||||
# Keep a negative/accepted pair visible even when a paper has many attempts.
|
||||
accepted = [row for row in attempts if row['latest_observed_status'] == 'accepted']
|
||||
first = accepted[0] if accepted else None
|
||||
other_venue = next((row for row in accepted if row['venue_id'] != first['venue_id']), None)
|
||||
pair = None
|
||||
for negative in attempts:
|
||||
if negative['historical_negative'] and first:
|
||||
candidate = first if first['venue_id'] != negative['venue_id'] else other_venue
|
||||
if candidate:
|
||||
pair = (negative, candidate)
|
||||
break
|
||||
if pair is None:
|
||||
for negative in attempts:
|
||||
if negative['historical_negative'] and first:
|
||||
candidate = first if first['submission_id'] != negative['submission_id'] else (
|
||||
accepted[1] if len(accepted) > 1 else None)
|
||||
if candidate:
|
||||
pair = (negative, candidate)
|
||||
break
|
||||
selected = {}
|
||||
for row in list(pair or ()) + attempts:
|
||||
if len(selected) < 20:
|
||||
selected.setdefault(row['submission_id'], dict(row))
|
||||
for row in selected.values():
|
||||
row['decisions'] = []
|
||||
for decision in connection.execute('''SELECT id,status,raw_value,evidence_url,
|
||||
decision_date,observed_at FROM decisions WHERE submission_id=?
|
||||
ORDER BY observed_at DESC,COALESCE(decision_date,observed_at) DESC,id DESC LIMIT 50''',
|
||||
(row['submission_id'],)):
|
||||
entry = dict(decision)
|
||||
sources = [dict(source) for source in connection.execute('''
|
||||
SELECT so.name,r.external_id,ds.source_decision_id FROM decision_sources ds
|
||||
JOIN source_records r ON r.id=ds.source_record_id JOIN sources so ON so.id=r.source_id
|
||||
WHERE ds.decision_id=? ORDER BY so.name,r.external_id LIMIT 101''', (entry['id'],))]
|
||||
entry['sources'] = sources[:100]
|
||||
entry['sources_truncated'] = len(sources) > 100
|
||||
row['decisions'].append(entry)
|
||||
row['decisions_truncated'] = row['decision_observations'] > len(row['decisions'])
|
||||
paper_id = attempts[0]['paper_id']
|
||||
return {'paper_id': paper_id,
|
||||
'title': connection.execute('SELECT title FROM papers WHERE id=?', (paper_id,)).fetchone()[0],
|
||||
'counts': _paper_counts(attempts), 'attempt_count': len(attempts),
|
||||
'attempts': list(selected.values()), 'attempts_truncated': len(attempts) > len(selected),
|
||||
'chronology': 'Not inferred from observed_at, edition year, or row order.'}
|
||||
|
||||
|
||||
def report(db=None, *, pg_service=None, pg_schema=None, examples=20):
|
||||
if not 0 <= examples <= 100:
|
||||
raise ValueError('Examples must be between 0 and 100')
|
||||
started_at = timestamp()
|
||||
connection = open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)
|
||||
if not pg_service:
|
||||
connection.row_factory = sqlite3.Row
|
||||
try:
|
||||
if pg_service:
|
||||
identity = connection.execute('SELECT current_database(),current_schema()').fetchone()
|
||||
database, schema = identity[0], identity[1]
|
||||
else:
|
||||
database, schema = str(Path(db).resolve()), None
|
||||
names = ['papers_with_submissions', 'papers_with_multiple_submissions',
|
||||
'papers_with_historical_negative', 'papers_with_current_known_acceptance',
|
||||
'papers_with_negative_and_acceptance_in_distinct_attempts',
|
||||
'papers_with_negative_and_acceptance_in_distinct_venues']
|
||||
totals = Counter({key: 0 for key in names})
|
||||
totals.update(submissions=0, decision_observations=0,
|
||||
attempts_with_historical_negative=0,
|
||||
attempts_with_historical_negative_and_acceptance=0)
|
||||
latest = Counter()
|
||||
sample_attempts = []
|
||||
|
||||
def consume(rows):
|
||||
if not rows:
|
||||
return
|
||||
counts = _paper_counts(rows)
|
||||
totals.update(counts)
|
||||
if counts['papers_with_negative_and_acceptance_in_distinct_attempts'] and len(sample_attempts) < examples:
|
||||
sample_attempts.append(rows)
|
||||
|
||||
current = []
|
||||
from paper_atlas.db.events import has_event_key
|
||||
attempts_sql = ATTEMPTS
|
||||
if has_event_key(connection):
|
||||
attempts_sql = attempts_sql.replace('e.year,e.track,', 'e.year,e.track,e.event_key,', 1).replace('e.year,e.track\n', 'e.year,e.track,e.event_key\n')
|
||||
for raw in connection.execute(attempts_sql):
|
||||
row = dict(raw)
|
||||
if current and row['paper_id'] != current[0]['paper_id']:
|
||||
consume(current)
|
||||
current = []
|
||||
current.append(row)
|
||||
totals['submissions'] += 1
|
||||
totals['decision_observations'] += row['decision_observations']
|
||||
totals['attempts_with_historical_negative'] += row['historical_negative']
|
||||
totals['attempts_with_historical_negative_and_acceptance'] += int(
|
||||
row['historical_negative'] and row['historical_acceptance'])
|
||||
latest[row['latest_observed_status']] += 1
|
||||
consume(current)
|
||||
# These are relationship checks, not a substitute for the final entity/FK audit.
|
||||
integrity = {}
|
||||
checks = {
|
||||
'submissions_without_source_links': '''SELECT count(*) FROM submissions s
|
||||
WHERE NOT EXISTS(SELECT 1 FROM submission_sources ss WHERE ss.submission_id=s.id)''',
|
||||
'decisions_without_source_links': '''SELECT count(*) FROM decisions d
|
||||
WHERE NOT EXISTS(SELECT 1 FROM decision_sources ds WHERE ds.decision_id=d.id)''',
|
||||
'submission_source_paper_mismatches': '''SELECT count(*) FROM submission_sources ss
|
||||
JOIN submissions s ON s.id=ss.submission_id JOIN source_records r ON r.id=ss.source_record_id
|
||||
WHERE s.paper_id<>r.paper_id''',
|
||||
'decision_source_paper_mismatches': '''SELECT count(*) FROM decision_sources ds
|
||||
JOIN decisions d ON d.id=ds.decision_id JOIN submissions s ON s.id=d.submission_id
|
||||
JOIN source_records r ON r.id=ds.source_record_id WHERE s.paper_id<>r.paper_id''',
|
||||
}
|
||||
for name, query in checks.items():
|
||||
integrity[name] = connection.execute(query).fetchone()[0]
|
||||
return {'report_started_at': started_at, 'checked_at': timestamp(),
|
||||
'engine': 'postgresql' if pg_service else 'sqlite', 'database': database,
|
||||
'schema': schema, 'pg_service': pg_service,
|
||||
'status': 'observed_history_audit_not_final_dedup', 'counts': dict(totals),
|
||||
'latest_attempt_status_counts': dict(sorted(latest.items())),
|
||||
'evidence_relationship_checks': integrity,
|
||||
'examples': [_example(connection, rows) for rows in sample_attempts],
|
||||
'example_limit': examples,
|
||||
'interpretation': [
|
||||
'Counts use existing canonical paper links, not independently verified real-world identity.',
|
||||
'Negative includes rejected, desk_rejected and withdrawn observations; these remain distinct decisions.',
|
||||
'Known acceptance follows the latest observed decision per attempt, matching query --outcome.',
|
||||
'Distinct attempts/venues require separate submission/venue IDs; aliases do not create new venues.',
|
||||
'Coexisting evidence does not prove rejection happened before acceptance. observed_at is collection time.',
|
||||
'No known acceptance is not proof of never accepted. Source signatures and completeness are not established here.',
|
||||
'Examples are bounded; counts include all attempts and observations in one read-only snapshot.',
|
||||
'This does not perform entity merging, final deduplication, or label namespace/type validation.',
|
||||
]}
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
def main(argv=None, *, prog=None):
|
||||
parser = argparse.ArgumentParser(prog=prog)
|
||||
database_arguments(parser)
|
||||
parser.add_argument('--examples', type=int, default=20)
|
||||
parser.add_argument('--out')
|
||||
args = parser.parse_args(argv)
|
||||
result = report(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema, examples=args.examples)
|
||||
text = json.dumps(result, ensure_ascii=False, indent=2)
|
||||
if args.out:
|
||||
path = Path(args.out)
|
||||
stage = path.with_suffix('.part')
|
||||
stage.write_text(text + '\n')
|
||||
stage.replace(path)
|
||||
else:
|
||||
print(text)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -39,7 +39,9 @@ def audit(db, inventory, out, *, pg_service=None, pg_schema=None):
|
||||
if json.loads(path.with_suffix('.manifest.json').read_text()) != manifest:
|
||||
raise ValueError('Collected manifest differs from inventory')
|
||||
count = 0
|
||||
for line in path.read_text().splitlines():
|
||||
# Unicode paragraph/line separators are valid inside JSON strings;
|
||||
# split physical byte lines, not Unicode text boundaries.
|
||||
for line in path.read_bytes().splitlines():
|
||||
row = json.loads(line)
|
||||
if any(row[key] != manifest[key] for key in ('source', 'venue', 'year', 'track')):
|
||||
raise ValueError('Record edition differs from inventory')
|
||||
|
||||
+6
-1
@@ -6,6 +6,8 @@ from paper_atlas import __version__
|
||||
|
||||
|
||||
COMMANDS = {
|
||||
('annotate',): ('paper_atlas.curation.annotations', {}, '管理独立的研究类型与自定义标签证据'),
|
||||
('curate','identifiers'): ('paper_atlas.curation.identifiers', {}, '按证据补充稳定标识并生成独立待导入文件'),
|
||||
('query',): ('paper_atlas.querying', {}, '查询论文、标签和投稿历史'),
|
||||
('coverage',): ('paper_atlas.audits.coverage', {}, '查看全部候选会议的覆盖情况'),
|
||||
('download',): ('paper_atlas.io.snapshots', {}, '下载并校验固定版本快照'),
|
||||
@@ -25,6 +27,8 @@ 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','event-identity'): ('paper_atlas.ingestion.event_identity', {}, '按核验计划增加同年会议事件身份'),
|
||||
('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
|
||||
('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
|
||||
('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
|
||||
@@ -35,6 +39,7 @@ COMMANDS = {
|
||||
('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'),
|
||||
('audit','arxiv-tags'): ('paper_atlas.audits.arxiv_tags', {}, '逐记录核验 arXiv 类别及出处完整性'),
|
||||
('audit','public-openreview'): ('paper_atlas.audits.public_openreview', {}, '核对公开负面决定与已有投稿历史'),
|
||||
('audit','histories'): ('paper_atlas.audits.histories', {}, '只读核对跨会议投稿历史及决定来源关系'),
|
||||
}
|
||||
|
||||
|
||||
@@ -47,7 +52,7 @@ def main(argv=None):
|
||||
leaf=commands.add_parser(route[0],help=help_text,add_help=False)
|
||||
else:
|
||||
if route[0] not in groups:
|
||||
group=commands.add_parser(route[0],help={'collect':'采集来源元数据','catalogue':'盘点历史目录','import':'导入数据库','maintain':'维护数据库','audit':'核验数据'}[route[0]])
|
||||
group=commands.add_parser(route[0],help={'collect':'采集来源元数据','catalogue':'盘点历史目录','import':'导入数据库','maintain':'维护数据库','audit':'核验数据','curate':'按证据整理元数据'}[route[0]])
|
||||
groups[route[0]]=group.add_subparsers(dest='operation',required=True)
|
||||
leaf=groups[route[0]].add_parser(route[1],help=help_text,add_help=False)
|
||||
leaf.set_defaults(handler=(route,module,kwargs))
|
||||
|
||||
@@ -88,8 +88,12 @@ 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']
|
||||
if edition.get('track_scope_evidence_url'):
|
||||
metadata['track_scope_evidence_url']=edition['track_scope_evidence_url']
|
||||
flagged=bool(re.search(r'\b(?:withdrawn|retracted)\b',row.get('publtype') or '',re.I))
|
||||
result={'source':'dblp','external_id':row['key'],'url':'https://dblp.org/rec/'+row['key']+'.html',
|
||||
'title':title,'abstract':None,'authors':authors,'identifiers':own,
|
||||
|
||||
@@ -38,6 +38,39 @@ BOOKS = {
|
||||
}
|
||||
|
||||
|
||||
# These are the historical primary series themselves, not co-located workshops.
|
||||
# Exact proceedings keys, original names, ordinals and event years must agree.
|
||||
HISTORICAL_WORKSHOP_TITLES = {
|
||||
**{('aistats', year): prefix + ordinal + ' International Workshop on Artificial Intelligence and Statistics'
|
||||
for year, ordinal, prefix in [
|
||||
(1995, 'Fifth', 'Learning from Data - '),
|
||||
(1997, 'Sixth', 'Proceedings of the '),
|
||||
(1999, 'Seventh', 'Proceedings of the '),
|
||||
(2001, 'Eighth', 'Proceedings of the '),
|
||||
(2003, 'Ninth', 'Proceedings of the '),
|
||||
(2005, 'Tenth', 'Proceedings of the ')]},
|
||||
**{('colt', year): 'Proceedings of the ' + ordinal + ' Annual Workshop on Computational Learning Theory'
|
||||
for year, ordinal in [(1988, 'First'), (1989, 'Second'), (1990, 'Third'), (1991, 'Fourth')]},
|
||||
}
|
||||
|
||||
|
||||
def historical_track(family, base, suffix, volume_key, year, title):
|
||||
if suffix or base.casefold() != family or volume_key != f'conf/{family}/{year}':
|
||||
return None
|
||||
prefix = HISTORICAL_WORKSHOP_TITLES.get((family, year))
|
||||
if (prefix and title.casefold().startswith(prefix.casefold() + ',')
|
||||
and re.search(r'\b' + str(year) + r'\b', title)):
|
||||
url = 'https://aistats.org/other.html' if family == 'aistats' else 'https://www.learningtheory.org/'
|
||||
return ('main', 'Historical primary series named Workshop; exact proceedings key, ordinal, title and event year verified.', url)
|
||||
if (family == 'colt' and year == 2003
|
||||
and title.startswith('Computational Learning Theory and Kernel Machines, 16th Annual Conference on Computational Learning Theory and 7th Kernel Workshop, COLT/Kernel 2003,')
|
||||
and title.endswith('Proceedings')):
|
||||
return ('proceedings_unspecified',
|
||||
'Combined COLT 2003 and Kernel Workshop volume; per-paper section absent from DBLP XML. Do not infer a main or workshop track.',
|
||||
'https://doi.org/10.1007/b12006')
|
||||
return None
|
||||
|
||||
|
||||
def scope(row,volume,through_year=2026):
|
||||
fields=row['fields'];books=fields.get('booktitle',[])
|
||||
if len(books)!=1:return None,'ambiguous_booktitle'
|
||||
@@ -75,7 +108,9 @@ def scope(row,volume,through_year=2026):
|
||||
if len(bibliographic)!=1 or not re.fullmatch(r'\d{4}',bibliographic[0]):return None,'ambiguous_bibliographic_year'
|
||||
if int(bibliographic[0])!=year and not re.search(r'\b'+str(year)+r'\b',title):return None,'unverified_year_difference'
|
||||
text=(suffix+' '+title).casefold();track='main'
|
||||
if 'findings' in text:track='findings'
|
||||
historical = historical_track(family, base, suffix, volume['key'], year, title)
|
||||
if historical:track=historical[0]
|
||||
elif 'findings' in text:track='findings'
|
||||
elif 'shared task' in text:track='shared_task'
|
||||
elif 'student research' in text or 'workshop' in text or 'wokshop' in text or base in ('ICCVW','EWSL'):track='workshop'
|
||||
elif ('demo track' in text and ('industrial track' in text or 'applied data science track' in text)):track='proceedings_unspecified'
|
||||
@@ -96,8 +131,18 @@ 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 historical:
|
||||
result['track_scope_note'] = historical[1]
|
||||
result['track_scope_evidence_url'] = historical[2]
|
||||
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.
|
||||
|
||||
@@ -27,11 +27,25 @@ RAW_ROOT = 'https://raw.githubusercontent.com/papercopilot/openreview/' + REVISI
|
||||
TREE_URL = 'https://api.github.com/repos/papercopilot/openreview/git/trees/' + REVISION + '?recursive=1'
|
||||
NEGATIVE = {'reject': 'rejected', 'rejected': 'rejected',
|
||||
'desk reject': 'desk_rejected', 'desk rejected': 'desk_rejected',
|
||||
'withdrawn': 'withdrawn', 'withdrawal': 'withdrawn'}
|
||||
'withdraw': 'withdrawn', 'withdrawn': 'withdrawn', 'withdrawal': 'withdrawn'}
|
||||
TRACKS = {'main': 'main', 'Position': 'position',
|
||||
'Datasets & Benchmarks': 'datasets_benchmarks'}
|
||||
WORKSHOP_REJECT_LABELS = {"aim-fm workshop @ neurips'24 reject",
|
||||
'coginterp @ neurips 2025 reject'}
|
||||
# Exact curated labels, tied to the year and raw workshop track. The official
|
||||
# venue directory establishes workshop identity, not an original decision signature.
|
||||
ICLR_WORKSHOP_LABELS = {
|
||||
'iclr 2024 workshop mefomo withdrawn submission':
|
||||
(2024, 'workshop_me_fomo', 'withdrawn', 'ME-FoMo'),
|
||||
'iclr 2024 workshop genai4dm withdrawn submission':
|
||||
(2024, 'workshop_genai4dm', 'withdrawn', 'GenAI4DM'),
|
||||
'iclr 2025 workshop agenticai reject':
|
||||
(2025, 'workshop_agenticai', 'rejected', 'AgenticAI'),
|
||||
'iclr 2025 workshop lmrl withdrawn submission':
|
||||
(2025, 'workshop_lmrl', 'withdrawn', 'LMRL'),
|
||||
'iclr 2026 workshop llm reasoning withdrawn submission':
|
||||
(2026, 'workshop_llm_reasoning', 'withdrawn', 'LLM_Reasoning'),
|
||||
}
|
||||
|
||||
|
||||
def download(url, limit):
|
||||
@@ -63,6 +77,20 @@ def normalize(row, family, year, provenance, include_workshops=False):
|
||||
"""Return a normalized assertion or a reason requiring separate review."""
|
||||
raw_status = row.get('status', '')
|
||||
status = negative_status(raw_status)
|
||||
workshop = ICLR_WORKSHOP_LABELS.get(' '.join(raw_status.casefold().split()))
|
||||
workshop_metadata = {}
|
||||
if workshop:
|
||||
expected_year, expected_track, status, group_name = workshop
|
||||
if family != 'iclr' or year != expected_year:
|
||||
raise ValueError('Workshop status label conflicts with source conference year')
|
||||
if row.get('track') != expected_track:
|
||||
return None, 'conflicting_workshop_track'
|
||||
if not include_workshops:
|
||||
return None, 'separate_workshop'
|
||||
workshop_metadata = {
|
||||
'workshop_venueid': f'ICLR.cc/{year}/Workshop/{group_name}',
|
||||
'workshop_identity_evidence_url': 'https://openreview.net/venue?id=ICLR.cc',
|
||||
}
|
||||
if status is None:
|
||||
if re.search(r'\bretract\w*\b', raw_status, re.I):
|
||||
return None, 'retracted'
|
||||
@@ -90,7 +118,9 @@ def normalize(row, family, year, provenance, include_workshops=False):
|
||||
return None, 'unmapped_venue'
|
||||
title = row.get('title')
|
||||
abstract = row.get('abstract')
|
||||
if not isinstance(title, str) or not title.strip():
|
||||
if title is None or (isinstance(title, str) and not title.strip()):
|
||||
return None, 'missing_title'
|
||||
if not isinstance(title, str):
|
||||
raise ValueError('Public submission lacks a title')
|
||||
if abstract is not None and not isinstance(abstract, str):
|
||||
raise ValueError('Abstract must be text or null')
|
||||
@@ -116,7 +146,8 @@ def normalize(row, family, year, provenance, include_workshops=False):
|
||||
'venue': venue, 'year': year, 'track': normalized_track,
|
||||
'status': status, 'raw_decision': raw_status, 'evidence_url': evidence,
|
||||
'observed_at': AS_OF, 'tags': tags,
|
||||
'metadata': {**provenance, 'raw_track': track, 'raw_status': raw_status,
|
||||
'metadata': {**provenance, **workshop_metadata,
|
||||
'raw_track': track, 'raw_status': raw_status,
|
||||
'evidence_kind': 'curated_public_status_assertion',
|
||||
'original_decision_signature_verified': False,
|
||||
'source_snapshot_upper_bound': AS_OF}}, None
|
||||
@@ -215,7 +246,7 @@ def _collect_files(out, selected, include_workshops, fetcher):
|
||||
name = f'public-openreview-{family}-{year}-{track}.jsonl'
|
||||
path = stage / name
|
||||
path.write_text(''.join(json.dumps(row, ensure_ascii=False) + '\n' for row in records))
|
||||
manifest = {'source': SOURCE, 'parser_version': 1, 'venue': records[0]['venue'],
|
||||
manifest = {'source': SOURCE, 'parser_version': 3, 'venue': records[0]['venue'],
|
||||
'year': year, 'track': track, 'count': len(records), 'status': 'partial',
|
||||
'file': str(out / name), 'file_sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
|
||||
'list_url': REPOSITORY + '/tree/' + REVISION, 'source_revision': REVISION,
|
||||
|
||||
@@ -7,7 +7,7 @@ import re
|
||||
import time
|
||||
from datetime import date
|
||||
from urllib.error import HTTPError
|
||||
from urllib.parse import parse_qs, unquote, urljoin, urlsplit
|
||||
from urllib.parse import parse_qs, unquote, urldefrag, urljoin, urlsplit
|
||||
|
||||
from paper_atlas.io.http import fetch
|
||||
from paper_atlas.io.html import Document, Node, clean
|
||||
@@ -147,7 +147,8 @@ def read_index(index,cache):
|
||||
links=set()
|
||||
for node in Document(html).root.walk():
|
||||
if node.tag!='a':continue
|
||||
link=urljoin(url,node.attrs.get('href',''));day=parse_qs(urlsplit(link).query).get('day',[])
|
||||
# Fragments select positions within a page, not separate HTTP resources.
|
||||
link=urldefrag(urljoin(url,node.attrs.get('href','')))[0];day=parse_qs(urlsplit(link).query).get('day',[])
|
||||
if index_identity(link)!=index_identity(url) or index_identity(url) is None or len(day)!=1:continue
|
||||
if day[0]!='all':
|
||||
try:valid=date.fromisoformat(day[0]).year==index['year']
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Evidence-bound identity preparation and independent type/tag annotations."""
|
||||
@@ -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) or not conn.execute('SELECT 1 FROM schema_migrations WHERE name=?', (MIGRATION,)).fetchone():
|
||||
raise RuntimeError('Run paper-atlas annotate init explicitly before adding annotations')
|
||||
|
||||
|
||||
def _row(row):
|
||||
return dict(zip(COLUMNS, (row[i] for i in range(len(COLUMNS)))))
|
||||
|
||||
|
||||
def _replay(conn, request_id, payload):
|
||||
old = conn.execute('SELECT * FROM annotation_events WHERE request_id=?', (request_id,)).fetchone()
|
||||
if old is None:
|
||||
return None
|
||||
old = _row(old)
|
||||
if any(old[k] != value for k, value in payload.items()):
|
||||
raise ValueError('request_id already identifies different annotation content')
|
||||
return old['id']
|
||||
|
||||
|
||||
def assert_annotation(conn, paper, kind, namespace, value, *, request_id, evidence_key,
|
||||
reason, actor, evidence_url=None, source_record_id=None):
|
||||
_begin_write(conn)
|
||||
_require(conn)
|
||||
paper = _id(paper, 'paper')
|
||||
if kind not in ('tag', 'research_type'):
|
||||
raise ValueError('kind must be tag or research_type')
|
||||
_text(namespace, 'namespace', 128)
|
||||
if not re.fullmatch(r'[a-z][a-z0-9_]*:[a-z][a-z0-9_.-]*', namespace):
|
||||
raise ValueError('namespace must have an explicit source:scope')
|
||||
payload = dict(original_paper_id=paper, kind=kind, namespace=namespace,
|
||||
value=_text(value, 'value'), request_id=_text(request_id, 'request_id', 128),
|
||||
source_record_id=source_record_id, evidence_key=_text(evidence_key, 'evidence_key', 256),
|
||||
reason=_text(reason, 'reason', 8192), actor=_text(actor, 'actor', 256),
|
||||
evidence_url=_url(evidence_url), retracts_id=None)
|
||||
if source_record_id is not None:
|
||||
_id(source_record_id, 'source_record_id')
|
||||
replay = _replay(conn, request_id, payload)
|
||||
if replay is not None:
|
||||
return replay
|
||||
if not conn.execute('SELECT 1 FROM papers WHERE id=?', (paper,)).fetchone():
|
||||
raise ValueError('Existing canonical paper required; no automatic identity merge')
|
||||
prefix = namespace.split(':', 1)[0]
|
||||
if prefix != 'custom' and source_record_id is None:
|
||||
raise ValueError('Platform annotations require their own source record')
|
||||
if source_record_id is not None:
|
||||
source = conn.execute('''SELECT r.paper_id,s.name FROM source_records r
|
||||
JOIN sources s ON s.id=r.source_id WHERE r.id=?''', (source_record_id,)).fetchone()
|
||||
if not source or source[0] != paper or (prefix != 'custom' and prefix != source[1]):
|
||||
raise ValueError('Annotation source record/namespace must belong to this paper and platform')
|
||||
payload.update(paper_id=paper, observed_at=timestamp())
|
||||
fields = tuple(payload)
|
||||
return inserted_id(conn, 'INSERT INTO annotation_events(' + ','.join(fields) + ') VALUES(' +
|
||||
','.join('?' for _ in fields) + ')', tuple(payload[k] for k in fields))
|
||||
|
||||
|
||||
def retract_annotation(conn, event, *, request_id, reason, actor, evidence_url=None, custom_only=False):
|
||||
_begin_write(conn)
|
||||
_require(conn)
|
||||
event = _id(event, 'event')
|
||||
payload = dict(request_id=_text(request_id, 'request_id', 128), retracts_id=event,
|
||||
reason=_text(reason, 'reason', 8192), actor=_text(actor, 'actor', 256), evidence_url=_url(evidence_url))
|
||||
target = conn.execute('SELECT * FROM annotation_events WHERE id=?', (event,)).fetchone()
|
||||
if target is None or target[-1] is not None:
|
||||
raise ValueError('Retraction must reference an existing assertion')
|
||||
target = _row(target)
|
||||
if custom_only and not target['namespace'].startswith('custom:'):
|
||||
raise ValueError('Manual CLI cannot retract platform evidence')
|
||||
replay = _replay(conn, request_id, payload)
|
||||
if replay is not None:
|
||||
return replay
|
||||
if conn.execute('SELECT 1 FROM annotation_events WHERE retracts_id=?', (event,)).fetchone():
|
||||
raise ValueError('Assertion already retracted; its history is retained')
|
||||
payload.update({k: target[k] for k in ('paper_id', 'original_paper_id', 'kind', 'namespace', 'value', 'source_record_id', 'evidence_key')})
|
||||
payload['observed_at'] = timestamp()
|
||||
fields = tuple(payload)
|
||||
return inserted_id(conn, 'INSERT INTO annotation_events(' + ','.join(fields) + ') VALUES(' +
|
||||
','.join('?' for _ in fields) + ')', tuple(payload[k] for k in fields))
|
||||
|
||||
|
||||
def observations(conn, paper, *, kind=None, history=False):
|
||||
if not available(conn):
|
||||
return []
|
||||
where = ['a.paper_id=?']; params = [_id(paper, 'paper')]
|
||||
if not history:
|
||||
where.append(ACTIVE)
|
||||
if kind is not None:
|
||||
if kind not in ('tag', 'research_type'):
|
||||
raise ValueError('Invalid annotation kind')
|
||||
where.append('a.kind=?'); params.append(kind)
|
||||
return [_row(r) for r in conn.execute('SELECT a.* FROM annotation_events a WHERE ' +
|
||||
' AND '.join(where) + ' ORDER BY a.id', params)]
|
||||
|
||||
|
||||
def main(argv=None, *, prog=None):
|
||||
parser = argparse.ArgumentParser(prog=prog, description='显式初始化、追加或撤回带证据的自定义标签和研究类型')
|
||||
actions = parser.add_subparsers(dest='action', required=True)
|
||||
for action in ('init', 'assert', 'retract', 'list'):
|
||||
p = actions.add_parser(action)
|
||||
database_arguments(p)
|
||||
if action in ('assert', 'retract'):
|
||||
for field in ('request-id', 'reason', 'actor'):
|
||||
p.add_argument('--' + field, required=True)
|
||||
p.add_argument('--evidence-url')
|
||||
if action in ('assert', 'list'):
|
||||
p.add_argument('--paper', type=int, required=True)
|
||||
p.add_argument('--kind', choices=('tag', 'research_type'), required=action == 'assert')
|
||||
if action == 'assert':
|
||||
for field in ('namespace', 'value', 'evidence-key'):
|
||||
p.add_argument('--' + field, required=True)
|
||||
p.add_argument('--source-record', type=int)
|
||||
if action == 'retract':
|
||||
p.add_argument('--event', type=int, required=True)
|
||||
if action == 'list':
|
||||
p.add_argument('--history', action='store_true')
|
||||
args = parser.parse_args(argv)
|
||||
if args.action == 'list':
|
||||
with closing(open_read_connection(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema)) as conn:
|
||||
result = observations(conn, args.paper, kind=args.kind, history=args.history)
|
||||
else:
|
||||
if args.action == 'assert' and not args.namespace.startswith('custom:'):
|
||||
parser.error('Manual CLI requires custom:* namespace; platform assertions use validated source APIs')
|
||||
with closing(open_store(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema).conn) as conn:
|
||||
try:
|
||||
if not getattr(conn, 'postgres', False):
|
||||
conn.commit()
|
||||
conn.execute('BEGIN IMMEDIATE')
|
||||
if args.action == 'init':
|
||||
result = {'initialized': initialize(conn)}
|
||||
else:
|
||||
kwargs = dict(request_id=args.request_id, reason=args.reason, actor=args.actor, evidence_url=args.evidence_url)
|
||||
event = (assert_annotation(conn, args.paper, args.kind, args.namespace, args.value,
|
||||
evidence_key=args.evidence_key, source_record_id=args.source_record, **kwargs)
|
||||
if args.action == 'assert' else retract_annotation(conn, args.event, custom_only=True, **kwargs))
|
||||
result = {'event_id': event}
|
||||
conn.commit()
|
||||
except BaseException:
|
||||
conn.rollback()
|
||||
raise
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
@@ -0,0 +1,212 @@
|
||||
"""Add reviewed stable identifiers without rewriting bibliographic source fields.
|
||||
|
||||
Hashes establish file integrity, not the truth of an identity assertion. Identity
|
||||
review and a fresh database preflight remain separate requirements.
|
||||
"""
|
||||
import argparse
|
||||
from datetime import datetime
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import tempfile
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
|
||||
SCOPE = 'input_and_evidence_integrity_only'
|
||||
MAX_METADATA_BYTES = 16 * 1024 * 1024
|
||||
|
||||
|
||||
def digest(data):
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def object_keys(value, required, optional=()):
|
||||
if not isinstance(value, dict) or not set(required) <= value.keys() or value.keys() - set(required) - set(optional):
|
||||
raise ValueError('Missing or unsupported plan fields')
|
||||
|
||||
|
||||
def json_bytes(value):
|
||||
return (json.dumps(value, ensure_ascii=False, separators=(',', ':'), allow_nan=False) + '\n').encode('utf-8')
|
||||
|
||||
|
||||
def decode(raw):
|
||||
def unique(pairs):
|
||||
result = {}
|
||||
for key, value in pairs:
|
||||
if key in result:
|
||||
raise ValueError('Duplicate JSON key: ' + key)
|
||||
result[key] = value
|
||||
return result
|
||||
return json.loads(raw, object_pairs_hook=unique,
|
||||
parse_constant=lambda value: (_ for _ in ()).throw(ValueError('Non-finite JSON value')))
|
||||
|
||||
|
||||
def metadata(path, expected=None):
|
||||
with Path(path).open('rb') as stream:
|
||||
raw = stream.read(MAX_METADATA_BYTES + 1)
|
||||
if len(raw) > MAX_METADATA_BYTES:
|
||||
raise ValueError('Metadata exceeds size limit')
|
||||
if expected is not None:
|
||||
check_hash(raw, expected)
|
||||
return raw, decode(raw)
|
||||
|
||||
|
||||
def check_hash(raw, expected):
|
||||
if not isinstance(expected, str) or not re.fullmatch('[0-9a-f]{64}', expected) or digest(raw) != expected:
|
||||
raise ValueError('SHA256 mismatch')
|
||||
|
||||
|
||||
def identifier(scheme, value):
|
||||
if not isinstance(value, str) or any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in value):
|
||||
raise ValueError('Invalid identifier')
|
||||
if scheme == 'arxiv':
|
||||
pattern = r'(?:\d{2}(?:0[1-9]|1[0-2])\.\d{4,5}|[a-zA-Z][a-zA-Z.-]*/\d{2}(?:0[1-9]|1[0-2])\d{3})'
|
||||
valid = re.fullmatch(pattern, value)
|
||||
elif scheme == 'doi':
|
||||
valid = re.fullmatch(r'10\.\d{4,9}/\S+', value) and value == value.lower()
|
||||
else:
|
||||
valid = False
|
||||
if not valid:
|
||||
raise ValueError('Unsupported or noncanonical identifier: ' + scheme)
|
||||
|
||||
|
||||
def prepare(plan_path, output):
|
||||
plan_raw, plan = metadata(plan_path)
|
||||
object_keys(plan, ('format_version', 'inputs', 'assertions'))
|
||||
if type(plan['format_version']) is not int or plan['format_version'] != 1:
|
||||
raise ValueError('Unsupported format version')
|
||||
if not isinstance(plan['inputs'], list) or not plan['inputs'] or not isinstance(plan['assertions'], list) or not plan['assertions']:
|
||||
raise ValueError('Inputs and assertions must be nonempty lists')
|
||||
output = Path(output).absolute()
|
||||
if output.exists():
|
||||
raise ValueError('Output directory must not already exist')
|
||||
if not output.parent.is_dir():
|
||||
raise ValueError('Output parent directory must exist')
|
||||
assertions = {}
|
||||
evidence_files = {}
|
||||
checked_evidence = {}
|
||||
for assertion in plan['assertions']:
|
||||
object_keys(assertion, ('source', 'external_id', 'record_sha256', 'identifiers', 'evidence'))
|
||||
if any(not isinstance(assertion[key], str) or not assertion[key] for key in ('source', 'external_id')):
|
||||
raise ValueError('Invalid source identity')
|
||||
key = (assertion['source'], assertion['external_id'])
|
||||
if key in assertions:
|
||||
raise ValueError('Duplicate identity assertion')
|
||||
ids = assertion['identifiers']
|
||||
if not isinstance(ids, dict) or not ids:
|
||||
raise ValueError('Identifiers must be a nonempty object')
|
||||
for scheme, value in ids.items():
|
||||
identifier(scheme, value)
|
||||
if not isinstance(assertion['evidence'], list) or not assertion['evidence']:
|
||||
raise ValueError('Evidence is required')
|
||||
references = []
|
||||
for evidence in assertion['evidence']:
|
||||
object_keys(evidence, ('file', 'sha256', 'url', 'checked_at', 'note'))
|
||||
if not all(isinstance(value, str) and value for value in evidence.values()):
|
||||
raise ValueError('Invalid evidence fields')
|
||||
url = urlsplit(evidence['url'])
|
||||
if url.scheme != 'https' or not url.hostname or url.username or url.password:
|
||||
raise ValueError('Evidence requires an HTTPS URL without credentials')
|
||||
if datetime.fromisoformat(evidence['checked_at'].replace('Z', '+00:00')).utcoffset() is None:
|
||||
raise ValueError('Evidence timestamp requires a timezone')
|
||||
evidence_key = (str(Path(evidence['file']).absolute()), evidence['sha256'])
|
||||
if evidence_key not in checked_evidence:
|
||||
checked_evidence[evidence_key] = metadata(evidence['file'], evidence['sha256'])[0]
|
||||
raw = checked_evidence[evidence_key]
|
||||
evidence_files[evidence['sha256']] = raw
|
||||
references.append({**evidence, 'file': 'evidence/' + evidence['sha256'] + '.json'})
|
||||
assertions[key] = (assertion, references)
|
||||
seen = set()
|
||||
applied = set()
|
||||
names = set()
|
||||
report = {'format_version': 1, 'plan_sha256': digest(plan_raw),
|
||||
'verification_scope': SCOPE, 'semantic_identity_verified': False,
|
||||
'database_written': False, 'files': [], 'assertions_applied': 0, 'identifiers_added': 0}
|
||||
with tempfile.TemporaryDirectory(prefix='.identifier-curation-', dir=output.parent) as temporary:
|
||||
staging = Path(temporary) / 'prepared'
|
||||
staging.mkdir()
|
||||
(staging / 'evidence').mkdir()
|
||||
(staging / 'plan.json').write_bytes(plan_raw)
|
||||
for sha, raw in evidence_files.items():
|
||||
(staging / 'evidence' / (sha + '.json')).write_bytes(raw)
|
||||
for entry in plan['inputs']:
|
||||
object_keys(entry, ('file', 'sha256', 'manifest_sha256'))
|
||||
path = Path(entry['file'])
|
||||
if path.suffix != '.jsonl' or path.name in names:
|
||||
raise ValueError('Inputs require distinct JSONL basenames')
|
||||
names.add(path.name)
|
||||
_, manifest = metadata(path.with_suffix('.manifest.json'), entry['manifest_sha256'])
|
||||
if not isinstance(manifest, dict) or not {'source','venue','year','track','count','list_url','file_sha256'} <= manifest.keys():
|
||||
raise ValueError('Complete source manifest required')
|
||||
if manifest['file_sha256'] != entry['sha256']:
|
||||
raise ValueError('Manifest and plan hashes differ')
|
||||
before = hashlib.sha256()
|
||||
after = hashlib.sha256()
|
||||
count = 0
|
||||
with path.open('rb') as source, (staging / path.name).open('wb') as target:
|
||||
for raw in source:
|
||||
before.update(raw)
|
||||
row = decode(raw)
|
||||
if any(row[field] != manifest[field] for field in ('source', 'venue', 'year', 'track')):
|
||||
raise ValueError('Record differs from source manifest')
|
||||
key = (row['source'], row['external_id'])
|
||||
if key in seen:
|
||||
raise ValueError('Duplicate source identity in input batch')
|
||||
seen.add(key)
|
||||
if key in assertions:
|
||||
assertion, references = assertions[key]
|
||||
check_hash(raw, assertion['record_sha256'])
|
||||
existing = row.get('identifiers', {})
|
||||
if existing is None:
|
||||
existing = {}
|
||||
if not isinstance(existing, dict):
|
||||
raise ValueError('Invalid existing identifiers')
|
||||
for scheme, value in assertion['identifiers'].items():
|
||||
if scheme in existing and existing[scheme] != value:
|
||||
raise ValueError('Existing identifier conflict')
|
||||
details = row.setdefault('metadata', {})
|
||||
if not isinstance(details, dict):
|
||||
raise ValueError('Invalid record metadata')
|
||||
history = details.setdefault('identifier_curation', [])
|
||||
if not isinstance(history, list):
|
||||
raise ValueError('Invalid curation history')
|
||||
history.append({'plan_sha256': digest(plan_raw), 'input_file_sha256': entry['sha256'],
|
||||
'record_sha256': digest(raw), 'previous_identifiers': dict(existing),
|
||||
'identifiers_asserted': assertion['identifiers'], 'evidence': references,
|
||||
'verification_scope': SCOPE})
|
||||
report['identifiers_added'] += len(assertion['identifiers'].keys() - existing.keys())
|
||||
row['identifiers'] = {**existing, **assertion['identifiers']}
|
||||
raw = json_bytes(row)
|
||||
applied.add(key)
|
||||
target.write(raw)
|
||||
after.update(raw)
|
||||
count += 1
|
||||
if before.hexdigest() != entry['sha256'] or not re.fullmatch('[0-9a-f]{64}', entry['sha256']):
|
||||
raise ValueError('Input SHA256 mismatch')
|
||||
if not count or type(manifest['count']) is not int or count != manifest['count']:
|
||||
raise ValueError('Manifest count mismatch')
|
||||
provenance = {'plan_sha256': digest(plan_raw), 'input_file_sha256': before.hexdigest(),
|
||||
'input_manifest_sha256': entry['manifest_sha256'], 'verification_scope': SCOPE}
|
||||
prepared = {**manifest, 'file': str(output / path.name), 'file_sha256': after.hexdigest(),
|
||||
'identifier_curation': provenance}
|
||||
(staging / path.with_suffix('.manifest.json').name).write_bytes(json_bytes(prepared))
|
||||
report['files'].append({'file': str(output / path.name), 'count': count,
|
||||
'input_sha256': before.hexdigest(), 'sha256': after.hexdigest()})
|
||||
if applied != assertions.keys():
|
||||
raise ValueError('Assertion target absent from input batch')
|
||||
report['assertions_applied'] = len(applied)
|
||||
(staging / 'curation-report.json').write_bytes(json_bytes(report))
|
||||
# All files are on the same filesystem. No database or input mutation.
|
||||
if output.exists():
|
||||
raise ValueError('Output directory appeared during preparation')
|
||||
staging.rename(output)
|
||||
return report
|
||||
|
||||
|
||||
def main(argv=None, *, prog=None):
|
||||
parser = argparse.ArgumentParser(prog=prog, description=__doc__)
|
||||
parser.add_argument('--plan', required=True, type=Path)
|
||||
parser.add_argument('--output', required=True, type=Path)
|
||||
args = parser.parse_args(argv)
|
||||
print(json.dumps(prepare(args.plan, args.output), ensure_ascii=False))
|
||||
@@ -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);
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Stable event identity within a venue/year/track, independent of paper identity."""
|
||||
import json
|
||||
import re
|
||||
|
||||
|
||||
def has_event_key(c):
|
||||
if getattr(c, 'postgres', False):
|
||||
return bool(c.execute("SELECT 1 FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='editions' AND column_name='event_key'").fetchone())
|
||||
return any(row[1] == 'event_key' for row in c.execute('PRAGMA table_info(editions)'))
|
||||
|
||||
|
||||
def validate_event_key(value):
|
||||
if value is not None and (not isinstance(value, str) or
|
||||
(value != '' and not re.fullmatch(r'[a-z0-9][a-z0-9._:-]{0,79}', value))):
|
||||
raise ValueError('Invalid conference event_key')
|
||||
return value
|
||||
|
||||
|
||||
def pinned_www_fall(source, metadata):
|
||||
from paper_atlas.collectors.dblp import XML_SHA
|
||||
b = metadata.get('bibliography') or {}
|
||||
urls = b.get('url') or []
|
||||
return (source == 'dblp' and metadata.get('snapshot_sha256') == XML_SHA
|
||||
and metadata.get('bibliographic_year') == 1995
|
||||
and str(metadata.get('proceedings_key', '')).startswith('conf/www/')
|
||||
and b.get('booktitle') == ['WWW (Fall)'] and len(urls) == 1
|
||||
and urls[0].split('#')[0] == 'db/conf/www/www1995f.html')
|
||||
|
||||
|
||||
def find_edition(store, venue_id, venue, year, track, record, external_id,
|
||||
event_key=None, *, create=True):
|
||||
c = store.conn
|
||||
validate_event_key(event_key)
|
||||
enabled = has_event_key(c) if getattr(store, 'event_identity', None) is None else store.event_identity
|
||||
known_multi = venue.upper() == 'WWW' and int(year) in (1994, 1995)
|
||||
source = 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()
|
||||
inferred = ('www-1995-fall' if known_multi and int(year) == 1995 and source
|
||||
and pinned_www_fall(source[0], json.loads(source[1])) else None)
|
||||
if inferred:
|
||||
if event_key is not None and event_key != inferred:
|
||||
raise ValueError('event_key conflicts with pinned WWW Fall source evidence')
|
||||
event_key = inferred
|
||||
if not enabled:
|
||||
if event_key or known_multi:
|
||||
raise ValueError('Explicit event identity migration required before this conference import')
|
||||
row = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue_id, year, track)).fetchone()
|
||||
if row or not create:
|
||||
return row[0] if row else None
|
||||
c.execute('INSERT INTO editions(venue_id,year,track) VALUES(?,?,?) ON CONFLICT DO NOTHING', (venue_id, year, track))
|
||||
return c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue_id, year, track)).fetchone()[0]
|
||||
anchored = c.execute('''SELECT DISTINCT e.id,e.event_key FROM editions e JOIN submissions s ON s.edition_id=e.id
|
||||
JOIN submission_sources ss ON ss.submission_id=s.id WHERE e.venue_id=? AND e.year=? AND e.track=?
|
||||
AND ss.source_record_id=? AND ss.source_submission_id=?''', (venue_id, year, track, record, external_id)).fetchall()
|
||||
if len(anchored) > 1:
|
||||
raise ValueError('Source record already has ambiguous conference events')
|
||||
if anchored:
|
||||
if event_key is not None and event_key != anchored[0][1]:
|
||||
raise ValueError('Cannot move an existing source to another conference event')
|
||||
event_key = anchored[0][1]
|
||||
if event_key is None:
|
||||
rows = c.execute('SELECT event_key FROM editions WHERE venue_id=? AND year=? AND track=?', (venue_id, year, track)).fetchall()
|
||||
if len(rows) > 1 or any(row[0] for row in rows):
|
||||
raise ValueError('Explicit event_key required; conference event is ambiguous')
|
||||
event_key = ''
|
||||
if known_multi and not event_key:
|
||||
raise ValueError('Explicit event_key required for same-year WWW events')
|
||||
row = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?', (venue_id, year, track, event_key)).fetchone()
|
||||
if row or not create:
|
||||
return row[0] if row else None
|
||||
c.execute('INSERT INTO editions(venue_id,year,track,event_key) VALUES(?,?,?,?) ON CONFLICT DO NOTHING', (venue_id, year, track, event_key))
|
||||
return c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?', (venue_id, year, track, event_key)).fetchone()[0]
|
||||
@@ -29,7 +29,8 @@ CREATE TABLE IF NOT EXISTS venues (
|
||||
CREATE TABLE IF NOT EXISTS editions (
|
||||
id INTEGER PRIMARY KEY, venue_id INTEGER NOT NULL REFERENCES venues(id),
|
||||
year INTEGER NOT NULL, track TEXT NOT NULL DEFAULT 'main',
|
||||
UNIQUE(venue_id,year,track)
|
||||
event_key TEXT NOT NULL DEFAULT '',
|
||||
UNIQUE(venue_id,year,track,event_key)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS venue_aliases (
|
||||
venue_id INTEGER NOT NULL REFERENCES venues(id), alias TEXT NOT NULL,
|
||||
|
||||
+35
-10
@@ -36,6 +36,8 @@ class Store:
|
||||
self.conn.execute('PRAGMA synchronous=NORMAL')
|
||||
self.conn.execute('PRAGMA cache_size=-32768')
|
||||
migrate_sources(self.conn)
|
||||
from paper_atlas.db.events import has_event_key
|
||||
self.event_identity = has_event_key(self.conn)
|
||||
|
||||
def source(self,name,url):
|
||||
self.conn.execute('INSERT INTO sources(name,homepage) VALUES(?,?) ON CONFLICT DO NOTHING',(name,url))
|
||||
@@ -55,6 +57,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))
|
||||
@@ -95,13 +100,19 @@ class Store:
|
||||
matches.add(old[0])
|
||||
key = normalized(title)
|
||||
title_matches = c.execute('SELECT id,authors_json,abstract FROM papers WHERE title_key=?',(key,)).fetchall()
|
||||
authors_set = author_keys(authors)
|
||||
for paper,old_authors,old_abstract in title_matches:
|
||||
overlap = authors_set & author_keys(json.loads(old_authors))
|
||||
# Cross-platform title matching needs independent author corroboration.
|
||||
same_source = c.execute('SELECT 1 FROM source_records WHERE paper_id=? AND source_id=? LIMIT 1',(paper,source_id)).fetchone()
|
||||
if not same_source and len(key)>=25 and (len(overlap)>=2 or (overlap and abstract and abstract.strip()==(old_abstract or '').strip())):
|
||||
matches.add(paper)
|
||||
# An existing identifier/source record anchors the entity. Title
|
||||
# similarity must not pull in another deposit or an extended version.
|
||||
if not matches:
|
||||
title_only_matches = set()
|
||||
authors_set = author_keys(authors)
|
||||
for paper,old_authors,old_abstract in title_matches:
|
||||
overlap = authors_set & author_keys(json.loads(old_authors))
|
||||
same_source = c.execute('SELECT 1 FROM source_records WHERE paper_id=? AND source_id=? LIMIT 1',(paper,source_id)).fetchone()
|
||||
if not same_source and len(key)>=25 and (len(overlap)>=2 or (overlap and abstract and abstract.strip()==(old_abstract or '').strip())):
|
||||
title_only_matches.add(paper)
|
||||
if len(title_only_matches)>1:
|
||||
raise ValueError('Ambiguous paper identity: multiple author-corroborated title matches; provide verified identifiers')
|
||||
matches = title_only_matches
|
||||
if matches:
|
||||
paper = min(matches)
|
||||
for other in sorted(matches-{paper}):
|
||||
@@ -142,15 +153,29 @@ class Store:
|
||||
tag=c.execute('SELECT id FROM tags WHERE namespace=? AND value=?',(namespace,value)).fetchone()[0]
|
||||
c.execute('INSERT INTO paper_tags VALUES(?,?,?,?) ON CONFLICT DO NOTHING',(paper,tag,record,evidence))
|
||||
|
||||
def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None,guard_status=False):
|
||||
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,event_key=None):
|
||||
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'))
|
||||
for alias in aliases or []:
|
||||
c.execute('INSERT INTO venue_aliases VALUES(?,?) ON CONFLICT DO NOTHING',(venue_id,alias))
|
||||
c.execute('INSERT INTO editions(venue_id,year,track) VALUES(?,?,?) ON CONFLICT DO NOTHING',(venue_id,int(year),track))
|
||||
edition = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?',(venue_id,int(year),track)).fetchone()[0]
|
||||
from paper_atlas.db.events import find_edition
|
||||
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 [])
|
||||
|
||||
@@ -37,9 +37,14 @@ def backfill(db,files, *, pg_service=None, pg_schema=None):
|
||||
seen.add(row['external_id'])
|
||||
if any(row.get(key)!=manifest.get(key) for key in ('source','venue','year','track')):
|
||||
raise ValueError('Evidence row differs from edition manifest')
|
||||
if 'event_key' in manifest and row.get('event_key')!=manifest['event_key']:
|
||||
raise ValueError('Event identity differs from manifest')
|
||||
require_track_record(row, manifest)
|
||||
record=c.execute('SELECT r.id,r.paper_id,r.metadata_json FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?',(row['source'],row['external_id'])).fetchone()
|
||||
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=?',(row['venue'],row['year'],row['track'])).fetchone()
|
||||
from paper_atlas.db.events import find_edition
|
||||
venue=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(row['venue'],)).fetchone()
|
||||
eid=find_edition(store,venue[0],row['venue'],row['year'],row['track'],record[0],row['external_id'],row.get('event_key'),create=False) if record and venue else None
|
||||
edition=(eid,) if eid is not None else None
|
||||
if not record or not edition:raise ValueError('Evidence backfill requires an already imported record and edition')
|
||||
if not c.execute('SELECT 1 FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id WHERE ss.source_record_id=? AND ss.source_submission_id=? AND s.edition_id=?',(record[0],row['external_id'],edition[0])).fetchone():
|
||||
raise ValueError('No existing submission to backfill')
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
"""Explicit, plan-bound edition event-key upgrade; never reimport submissions."""
|
||||
import argparse
|
||||
from contextlib import closing
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
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, OPTIONAL_TABLES, quote
|
||||
|
||||
MARKER = 'edition_event_identity_v1:'
|
||||
|
||||
|
||||
def digest(value):
|
||||
return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
||||
|
||||
|
||||
def values(rows):
|
||||
return [[r[i] for i in range(len(r))] for r in rows]
|
||||
|
||||
|
||||
def exact_counts(c, *, page_size=5000, progress=None):
|
||||
"""Count every row in the same snapshot through bounded primary-key pages."""
|
||||
if not isinstance(page_size, int) or page_size < 1:
|
||||
raise ValueError('Positive count page size required')
|
||||
result = {}
|
||||
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):
|
||||
columns = [r[0] for r in c.execute("SELECT a.attname FROM pg_constraint pc JOIN pg_attribute a ON a.attrelid=pc.conrelid AND a.attnum=ANY(pc.conkey) WHERE pc.conrelid=?::regclass AND pc.contype='p' ORDER BY array_position(pc.conkey,a.attnum)", (table,))]
|
||||
else:
|
||||
columns = [r[1] for r in sorted(c.execute('PRAGMA table_info('+quote(table)+')'), key=lambda r:r[5]) if r[5]]
|
||||
if not columns:
|
||||
raise ValueError('Primary key required for exact count: '+table)
|
||||
fields = ','.join(quote(column) for column in columns)
|
||||
last = None; count = 0; pages = 0
|
||||
if progress:
|
||||
progress({'table':table, 'rows_counted':0, 'complete':False})
|
||||
while True:
|
||||
where = (' WHERE ('+fields+') > ('+','.join('?' for _ in columns)+')') if last is not None else ''
|
||||
batch = values(c.execute('SELECT '+fields+' FROM '+quote(table)+where+' ORDER BY '+fields+' LIMIT ?', (*last, page_size) if last is not None else (page_size,)).fetchall())
|
||||
count += len(batch); pages += 1
|
||||
complete = len(batch) < page_size
|
||||
if progress and (complete or pages % 20 == 0):
|
||||
progress({'table':table, 'rows_counted':count, 'complete':complete})
|
||||
if complete:
|
||||
break
|
||||
last = batch[-1]
|
||||
result[table] = count
|
||||
return result
|
||||
|
||||
|
||||
def inspect(c, *, progress=None):
|
||||
enabled = has_event_key(c)
|
||||
if getattr(c, 'postgres', False):
|
||||
columns = [r[0] for r in c.execute("SELECT column_name FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='editions' ORDER BY ordinal_position")]
|
||||
unique = [r[0] for r in c.execute("SELECT pg_get_constraintdef(oid) FROM pg_constraint WHERE conrelid='editions'::regclass AND contype='u'")]
|
||||
expected_unique = ['UNIQUE (venue_id, year, track, event_key)' if enabled else 'UNIQUE (venue_id, year, track)']
|
||||
else:
|
||||
columns = [r[1] for r in c.execute('PRAGMA table_info(editions)')]
|
||||
indexes = list(c.execute('PRAGMA index_list(editions)'))
|
||||
unique = [tuple(r[2] for r in c.execute('PRAGMA index_info('+quote(index[1])+')')) for index in indexes if index[2]]
|
||||
expected_unique = [('venue_id','year','track','event_key') if enabled else ('venue_id','year','track')]
|
||||
if len(indexes) != 1:
|
||||
raise ValueError('Unexpected edition indexes; require a reviewed migration')
|
||||
if columns != ['id','venue_id','year','track'] + (['event_key'] if enabled else []) or unique != expected_unique:
|
||||
raise ValueError('Unexpected edition schema; require a reviewed migration')
|
||||
rows = values(c.execute('SELECT id,venue_id,year,track FROM editions ORDER BY id'))
|
||||
keys = values(c.execute('SELECT id,event_key FROM editions ORDER BY id')) if enabled else [[r[0], ''] for r in rows]
|
||||
counts = exact_counts(c, progress=progress)
|
||||
targets = values(c.execute("SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym='WWW' AND e.year=1995 AND e.track='main' ORDER BY e.id"))
|
||||
assignments = []
|
||||
for (eid,) in targets:
|
||||
current_key = dict(keys)[eid]
|
||||
if current_key:
|
||||
continue
|
||||
evidence = values(c.execute('''SELECT s.id,src.name,r.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=? ORDER BY s.id,r.id''', (eid,)))
|
||||
count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (eid,)).fetchone()[0]
|
||||
supported = {r[0] for r in evidence if pinned_www_fall(r[1], json.loads(r[4]))}
|
||||
if not count or len(supported) != count:
|
||||
raise ValueError('Existing unlabelled WWW1995 edition lacks complete pinned Fall evidence')
|
||||
assignments.append({'edition_id':eid, 'event_key':'www-1995-fall', 'submissions':count,
|
||||
'source_evidence_sha256':digest(evidence)})
|
||||
# Protect all rows in the affected attempt history, including non-DBLP evidence.
|
||||
target_ids = [r[0] for r in targets]
|
||||
protected = {}
|
||||
for table,sql in {
|
||||
'submissions': 'SELECT s.* FROM submissions s WHERE s.edition_id IN ({}) ORDER BY s.id',
|
||||
'decisions': 'SELECT d.* FROM decisions d JOIN submissions s ON s.id=d.submission_id WHERE s.edition_id IN ({}) ORDER BY d.id',
|
||||
'submission_sources': 'SELECT ss.* FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id WHERE s.edition_id IN ({}) ORDER BY ss.submission_id,ss.source_record_id,ss.source_submission_id',
|
||||
'decision_sources': 'SELECT ds.* FROM decision_sources ds JOIN decisions d ON d.id=ds.decision_id JOIN submissions s ON s.id=d.submission_id WHERE s.edition_id IN ({}) ORDER BY ds.decision_id,ds.source_record_id,ds.source_decision_id',
|
||||
'source_records': 'SELECT DISTINCT r.* FROM source_records r JOIN submission_sources ss ON ss.source_record_id=r.id JOIN submissions s ON s.id=ss.submission_id WHERE s.edition_id IN ({}) ORDER BY r.id',
|
||||
}.items():
|
||||
records = values(c.execute(sql.format(','.join('?' for _ in target_ids)), target_ids)) if target_ids else []
|
||||
protected[table] = {'count':len(records), 'sha256':digest(records)}
|
||||
return {'event_column':enabled, 'editions_sha256':digest(rows), 'edition_ids':[r[0] for r in rows],
|
||||
'event_keys':keys, 'counts':counts, 'assignments':assignments, 'protected':protected}
|
||||
|
||||
|
||||
def database_identity(c, db=None):
|
||||
if getattr(c, 'postgres', False):
|
||||
r = c.execute('SELECT current_database(),current_schema()').fetchone()
|
||||
return {'engine':'postgres', 'database':r[0], 'schema':r[1]}
|
||||
return {'engine':'sqlite', 'path':str(Path(db).resolve())}
|
||||
|
||||
|
||||
def plan(db=None, *, pg_service=None, pg_schema=None, progress=None):
|
||||
with closing(open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)) as c:
|
||||
return {'format':1, 'database':database_identity(c, db), 'before':inspect(c, progress=progress)}
|
||||
|
||||
|
||||
def apply_plan(c, proposal, *, progress=None):
|
||||
before = proposal['before']
|
||||
if proposal.get('format') != 1:
|
||||
raise ValueError('Unsupported event identity plan')
|
||||
if getattr(c, 'postgres', False):
|
||||
c.execute('LOCK TABLE editions IN ACCESS EXCLUSIVE MODE')
|
||||
current = inspect(c, progress=progress)
|
||||
signature = digest(proposal)
|
||||
marker = MARKER + signature
|
||||
if c.execute('SELECT 1 FROM schema_migrations WHERE name=?', (marker,)).fetchone():
|
||||
expected_keys = dict(before['event_keys'])
|
||||
expected_keys.update({a['edition_id']:a['event_key'] for a in before['assignments']})
|
||||
if (not current['event_column'] or current['editions_sha256'] != before['editions_sha256']
|
||||
or current['protected'] != before['protected'] or current['counts'] != before['counts']
|
||||
or dict(current['event_keys']) != expected_keys):
|
||||
raise ValueError('Applied plan no longer matches current data; generate a new plan')
|
||||
return {'status':'already_applied', 'updated_editions':0, 'after':current}
|
||||
if current != before:
|
||||
raise ValueError('Event identity plan is stale; generate a new plan')
|
||||
if not current['event_column']:
|
||||
if getattr(c, 'postgres', False):
|
||||
constraints = c.execute("SELECT conname,pg_get_constraintdef(oid) FROM pg_constraint WHERE conrelid='editions'::regclass AND contype='u'").fetchall()
|
||||
matches = [name for name,definition in constraints if definition == 'UNIQUE (venue_id, year, track)']
|
||||
if len(matches) != 1 or len(constraints) != 1:
|
||||
raise ValueError('Unexpected legacy edition uniqueness')
|
||||
c.execute("ALTER TABLE editions ADD COLUMN event_key TEXT COLLATE \"C\" NOT NULL DEFAULT ''")
|
||||
c.execute('ALTER TABLE editions DROP CONSTRAINT '+quote(matches[0]))
|
||||
c.execute('ALTER TABLE editions ADD CONSTRAINT editions_event_identity_unique UNIQUE(venue_id,year,track,event_key)')
|
||||
else:
|
||||
unique = [tuple(r[2] for r in c.execute('PRAGMA index_info('+quote(index[1])+')'))
|
||||
for index in c.execute('PRAGMA index_list(editions)') if index[2]]
|
||||
if unique != [('venue_id','year','track')]:
|
||||
raise ValueError('Unexpected legacy edition uniqueness')
|
||||
c.execute("CREATE TABLE editions_event_upgrade(id INTEGER PRIMARY KEY,venue_id INTEGER NOT NULL REFERENCES venues(id),year INTEGER NOT NULL,track TEXT NOT NULL DEFAULT 'main',event_key TEXT NOT NULL DEFAULT '',UNIQUE(venue_id,year,track,event_key))")
|
||||
c.execute('INSERT INTO editions_event_upgrade(id,venue_id,year,track) SELECT id,venue_id,year,track FROM editions')
|
||||
c.execute('DROP TABLE editions')
|
||||
c.execute('ALTER TABLE editions_event_upgrade RENAME TO editions')
|
||||
for item in before['assignments']:
|
||||
result = c.execute("UPDATE editions SET event_key=? WHERE id=? AND event_key=''", (item['event_key'],item['edition_id']))
|
||||
if result.rowcount != 1:
|
||||
raise ValueError('Edition assignment changed during upgrade')
|
||||
c.execute('INSERT INTO schema_migrations(name) VALUES(?)', (marker,))
|
||||
after = inspect(c, progress=progress)
|
||||
expected_keys = dict(before['event_keys'])
|
||||
expected_keys.update({a['edition_id']:a['event_key'] for a in before['assignments']})
|
||||
if (after['editions_sha256'] != before['editions_sha256'] or after['protected'] != before['protected']
|
||||
or after['counts'] != before['counts'] or dict(after['event_keys']) != expected_keys):
|
||||
raise ValueError('Event upgrade changed protected data')
|
||||
if not getattr(c, 'postgres', False) and c.execute('PRAGMA foreign_key_check').fetchall():
|
||||
raise ValueError('Foreign key violation after event upgrade')
|
||||
return {'status':'applied', 'updated_editions':len(before['assignments']), 'after':after}
|
||||
|
||||
|
||||
def upgrade(proposal, db=None, *, pg_service=None, pg_schema=None, progress=None):
|
||||
if pg_service:
|
||||
from paper_atlas.postgres.writer import WriteConnection
|
||||
with closing(WriteConnection(pg_service, pg_schema)) as c:
|
||||
if proposal['database'] != database_identity(c):
|
||||
raise ValueError('Plan targets a different database')
|
||||
try:
|
||||
result = apply_plan(c, proposal, progress=progress); c.commit(); return result
|
||||
except BaseException:
|
||||
c.rollback(); raise
|
||||
require_sqlite_writable(db)
|
||||
if not Path(db).is_file():
|
||||
raise ValueError('Existing SQLite sample required')
|
||||
with closing(sqlite3.connect(db, timeout=30)) as c:
|
||||
if proposal['database'] != database_identity(c, db):
|
||||
raise ValueError('Plan targets a different database')
|
||||
c.execute('PRAGMA foreign_keys=OFF') # Outside the transaction; no cascading deletes.
|
||||
try:
|
||||
c.execute('BEGIN IMMEDIATE')
|
||||
result = apply_plan(c, proposal, progress=progress); c.commit(); return result
|
||||
except BaseException:
|
||||
c.rollback(); raise
|
||||
finally:
|
||||
c.execute('PRAGMA foreign_keys=ON')
|
||||
|
||||
|
||||
def main(argv=None, *, prog=None):
|
||||
p = argparse.ArgumentParser(prog=prog); database_arguments(p)
|
||||
p.add_argument('--out', required=True)
|
||||
p.add_argument('--apply-plan'); p.add_argument('--plan-sha256')
|
||||
args = p.parse_args(argv)
|
||||
if bool(args.apply_plan) != bool(args.plan_sha256):
|
||||
p.error('--apply-plan and --plan-sha256 must be supplied together')
|
||||
target = Path(args.out)
|
||||
if target.exists():
|
||||
raise ValueError('Refusing to overwrite an existing plan/report')
|
||||
progress = lambda event: print(json.dumps({'phase':'exact_row_count', **event}), file=sys.stderr, flush=True)
|
||||
if args.apply_plan:
|
||||
raw = Path(args.apply_plan).read_bytes()
|
||||
if hashlib.sha256(raw).hexdigest() != args.plan_sha256:
|
||||
raise ValueError('Plan file SHA256 differs from approved bytes')
|
||||
result = upgrade(json.loads(raw), args.db, pg_service=args.pg_service, pg_schema=args.pg_schema, progress=progress)
|
||||
else:
|
||||
result = plan(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema, progress=progress)
|
||||
target.write_text(json.dumps(result, ensure_ascii=False, indent=2)+'\n')
|
||||
print(json.dumps({'out':str(target), 'sha256':hashlib.sha256(target.read_bytes()).hexdigest(),
|
||||
'status':result.get('status','planned')}), flush=True)
|
||||
@@ -0,0 +1,204 @@
|
||||
"""Evidence-bound repair of two existing IJCAI/ECAI edition memberships."""
|
||||
import argparse
|
||||
from contextlib import closing
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from paper_atlas.db.backend import database_arguments, open_read_connection, open_store
|
||||
from paper_atlas.db.submissions import timestamp
|
||||
from paper_atlas.collectors.dblp import XML_SHA
|
||||
|
||||
YEARS = (2018, 2022)
|
||||
TARGET = 'ECAI+IJCAI'
|
||||
MEMBERS = {'ECAI', 'IJCAI'}
|
||||
EVIDENCE = 'https://www.eurai.org/ecai/history'
|
||||
MARKER = 'joint_ijcai_ecai_2018_2022_v1'
|
||||
|
||||
|
||||
def _digest(c, sql, params=()):
|
||||
sha = hashlib.sha256(); count = 0
|
||||
for row in c.execute(sql, params):
|
||||
sha.update((json.dumps([row[i] for i in range(len(row))], ensure_ascii=False,
|
||||
separators=(',', ':')) + '\n').encode())
|
||||
count += 1
|
||||
if count > 100000:
|
||||
raise ValueError('Joint repair evidence scope exceeded bounded row limit')
|
||||
return {'rows': count, 'sha256': sha.hexdigest()}
|
||||
|
||||
|
||||
def fingerprints(c, ids):
|
||||
marks = ','.join('?' for _ in ids)
|
||||
submissions = 'SELECT id FROM submissions WHERE edition_id IN (' + marks + ')'
|
||||
papers = 'SELECT paper_id FROM submissions WHERE edition_id IN (' + marks + ')'
|
||||
decisions = 'SELECT id FROM decisions WHERE submission_id IN (' + submissions + ')'
|
||||
checks = {
|
||||
'papers': ('id IN (' + papers + ')', 'id'),
|
||||
'identifiers': ('paper_id IN (' + papers + ')', 'scheme,value'),
|
||||
'source_records': ('paper_id IN (' + papers + ')', 'id'),
|
||||
'paper_tags': ('paper_id IN (' + papers + ')', 'paper_id,tag_id,evidence_key'),
|
||||
'submissions': ('edition_id IN (' + marks + ')', 'id'),
|
||||
'decisions': ('submission_id IN (' + submissions + ')', 'id'),
|
||||
'submission_sources': ('submission_id IN (' + submissions + ')', 'submission_id,source_record_id,source_submission_id'),
|
||||
'submission_identifiers': ('edition_id IN (' + marks + ')', 'edition_id,scheme,value'),
|
||||
'decision_sources': ('decision_id IN (' + decisions + ')', 'decision_id,source_record_id,source_decision_id'),
|
||||
'edition_sources': ('edition_id IN (' + marks + ')', 'edition_id,source_id,list_url'),
|
||||
}
|
||||
return {table: _digest(c, 'SELECT * FROM ' + table + ' WHERE ' + where + ' ORDER BY ' + order, ids)
|
||||
for table, (where, order) in checks.items()}
|
||||
|
||||
|
||||
def inspect(c):
|
||||
target = c.execute('SELECT id FROM venues WHERE acronym=?', (TARGET,)).fetchall()
|
||||
if len(target) != 1:
|
||||
raise ValueError('Exactly one existing canonical joint venue required')
|
||||
target = target[0][0]
|
||||
aliases = {r[0] for r in c.execute('SELECT alias FROM venue_aliases WHERE venue_id=?', (target,))}
|
||||
if aliases != MEMBERS:
|
||||
raise ValueError('Joint target must already have exactly ECAI and IJCAI aliases')
|
||||
editions = []
|
||||
for year in YEARS:
|
||||
rows = c.execute('''SELECT e.id,e.venue_id,v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id
|
||||
WHERE v.acronym IN ('IJCAI','ECAI+IJCAI') AND e.year=? AND e.track='main' ''', (year,)).fetchall()
|
||||
if len(rows) != 1:
|
||||
raise ValueError('Expected one existing edition per year; missing or colliding editions require separate review')
|
||||
edition, venue, acronym = (rows[0][i] for i in range(3))
|
||||
count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (edition,)).fetchone()[0]
|
||||
if not 0 < count <= 5000:
|
||||
raise ValueError('Existing edition must have a nonempty bounded submission scope')
|
||||
# Every existing submission must be supported by its own pinned DBLP
|
||||
# volume record. No title/author matching, new attempts or text writes.
|
||||
evidence = c.execute('''SELECT s.id,r.external_id,r.metadata_json FROM submissions s
|
||||
JOIN submission_sources ss ON ss.submission_id=s.id
|
||||
JOIN source_records r ON r.id=ss.source_record_id JOIN sources src ON src.id=r.source_id
|
||||
WHERE s.edition_id=? AND src.name='dblp' ORDER BY s.id,r.id''', (edition,)).fetchall()
|
||||
seen = set()
|
||||
for row in evidence:
|
||||
sid, external, metadata = (row[i] for i in range(3))
|
||||
meta = json.loads(metadata)
|
||||
bibliography = meta.get('bibliography') or {}
|
||||
if (meta.get('snapshot_sha256') != XML_SHA
|
||||
or meta.get('proceedings_key') != 'conf/ijcai/' + str(year)
|
||||
or meta.get('bibliographic_year') != year
|
||||
or bibliography.get('booktitle') not in (['IJCAI'], ['IJCAI/ECAI'])
|
||||
or not external.startswith('conf/ijcai/')):
|
||||
raise ValueError('Source evidence outside the exact verified IJCAI/ECAI volume')
|
||||
seen.add(sid)
|
||||
if len(seen) != count:
|
||||
raise ValueError('Every submission must have matching original volume evidence')
|
||||
editions.append({'id': edition, 'year': year, 'track': 'main', 'from_venue_id': venue,
|
||||
'from_venue': acronym, 'target_venue_id': target, 'submissions': count,
|
||||
'evidence_url': EVIDENCE, 'proceedings_key': 'conf/ijcai/' + str(year)})
|
||||
ids = tuple(e['id'] for e in editions)
|
||||
return {'format_version': 1, 'kind': MARKER, 'checked_at': timestamp(), 'editions': editions,
|
||||
'content_fingerprints': fingerprints(c, ids), 'target_aliases': sorted(aliases),
|
||||
'status': 'already_joint' if all(e['from_venue_id'] == target for e in editions) else 'ready',
|
||||
'scope': 'Only these two existing edition venue references; no source text, paper, submission, decision, track, identifier or tag change.'}
|
||||
|
||||
|
||||
def validate_plan(expected):
|
||||
if not isinstance(expected, dict):
|
||||
raise ValueError('Repair plan must be a JSON object')
|
||||
if set(expected) != {'format_version','kind','checked_at','editions','content_fingerprints','target_aliases','status','scope','database'}:
|
||||
raise ValueError('Unexpected repair plan fields')
|
||||
if (type(expected['format_version']) is not int or expected['format_version'] != 1
|
||||
or expected['kind'] != MARKER or expected['status'] not in ('ready','already_joint')):
|
||||
raise ValueError('Unsupported repair plan')
|
||||
if (not isinstance(expected['database'],dict) or not isinstance(expected['content_fingerprints'],dict)
|
||||
or not isinstance(expected['editions'],list) or len(expected['editions']) != 2
|
||||
or expected['target_aliases'] != sorted(MEMBERS)):
|
||||
raise ValueError('Invalid repair plan structure')
|
||||
for edition, year in zip(expected['editions'], YEARS):
|
||||
if not isinstance(edition,dict) or edition.get('year') != year:
|
||||
raise ValueError('Repair plan must name the two exact ordered years')
|
||||
for key in ('id','year','from_venue_id','target_venue_id','submissions'):
|
||||
if type(edition.get(key)) is not int or edition[key] <= 0:
|
||||
raise ValueError('Repair plan IDs/counts must be positive integers')
|
||||
|
||||
|
||||
def apply_plan(c, expected):
|
||||
validate_plan(expected)
|
||||
actual = inspect(c)
|
||||
if actual['content_fingerprints'] != expected['content_fingerprints'] or actual['target_aliases'] != expected['target_aliases']:
|
||||
raise ValueError('Stale repair plan: protected content or aliases changed')
|
||||
if len(expected['editions']) != 2:
|
||||
raise ValueError('Exactly two editions required')
|
||||
for old, now in zip(expected['editions'], actual['editions']):
|
||||
if set(old) != set(now):
|
||||
raise ValueError('Unexpected edition fields')
|
||||
for key in ('id','year','track','target_venue_id','submissions','evidence_url','proceedings_key'):
|
||||
if old[key] != now[key]:
|
||||
raise ValueError('Stale repair plan: edition differs')
|
||||
if now['from_venue_id'] != old['from_venue_id'] and now['from_venue_id'] != now['target_venue_id']:
|
||||
raise ValueError('Edition membership differs from plan')
|
||||
if actual['status'] == 'already_joint':
|
||||
return {**actual, 'status':'already_joint', 'updated_editions':0}
|
||||
for edition in actual['editions']:
|
||||
if edition['from_venue_id'] == edition['target_venue_id']:
|
||||
continue
|
||||
changed = c.execute('''UPDATE editions SET venue_id=? WHERE id=? AND venue_id=? AND year=? AND track=?''',
|
||||
(edition['target_venue_id'],edition['id'],edition['from_venue_id'],edition['year'],edition['track']))
|
||||
if changed.rowcount != 1:
|
||||
raise ValueError('Concurrent edition change; entire repair must roll back')
|
||||
after = inspect(c)
|
||||
if after['status'] != 'already_joint' or after['content_fingerprints'] != actual['content_fingerprints']:
|
||||
raise ValueError('Postcheck changed protected content; entire repair must roll back')
|
||||
c.execute('INSERT INTO schema_migrations(name) VALUES(?) ON CONFLICT DO NOTHING', (MARKER,))
|
||||
return {**after, 'status':'applied', 'updated_editions':sum(e['from_venue_id'] != e['target_venue_id'] for e in actual['editions'])}
|
||||
|
||||
|
||||
def database_identity(c, db):
|
||||
if getattr(c, 'postgres', False):
|
||||
row = c.execute('SELECT current_database(),current_schema()').fetchone()
|
||||
return {'engine':'postgresql','database':row[0],'schema':row[1]}
|
||||
return {'engine':'sqlite','path':str(Path(db).resolve())}
|
||||
|
||||
|
||||
def run(db, out, *, pg_service=None, pg_schema=None, plan=None, plan_sha256=None):
|
||||
out = Path(out).resolve()
|
||||
if not out.parent.is_dir():
|
||||
raise ValueError('Report parent directory must already exist')
|
||||
if out.exists():
|
||||
raise ValueError('Preserve existing evidence report; choose a new output path')
|
||||
expected = None
|
||||
if plan:
|
||||
raw = Path(plan).read_bytes()
|
||||
if hashlib.sha256(raw).hexdigest() != plan_sha256:
|
||||
raise ValueError('Repair plan hash differs before opening database')
|
||||
expected = json.loads(raw)
|
||||
validate_plan(expected)
|
||||
elif plan_sha256:
|
||||
raise ValueError('Plan hash requires an existing plan')
|
||||
if db and out == Path(db).resolve():
|
||||
raise ValueError('Report must not overwrite database')
|
||||
writer = open_store(db, pg_service=pg_service, pg_schema=pg_schema) if expected is not None else None
|
||||
c = writer.conn if writer else open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)
|
||||
with closing(c):
|
||||
try:
|
||||
if expected is not None and not getattr(c, 'postgres', False):
|
||||
c.commit(); c.execute('BEGIN IMMEDIATE')
|
||||
identity = database_identity(c, db)
|
||||
if expected is not None and expected['database'] != identity:
|
||||
raise ValueError('Plan belongs to another database/schema')
|
||||
report = apply_plan(c, expected) if expected is not None else inspect(c)
|
||||
report['database'] = identity
|
||||
if expected is not None:
|
||||
report['plan_sha256'] = plan_sha256
|
||||
if expected is not None:
|
||||
c.commit()
|
||||
except BaseException:
|
||||
if expected is not None: c.rollback()
|
||||
raise
|
||||
out.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
|
||||
print(json.dumps({'status':report['status'],'editions':len(report['editions']),
|
||||
'updated_editions':report.get('updated_editions',0),'report':str(out)}))
|
||||
return report
|
||||
|
||||
|
||||
def main(argv=None, *, prog=None):
|
||||
parser = argparse.ArgumentParser(prog=prog); database_arguments(parser)
|
||||
parser.add_argument('--out', required=True)
|
||||
parser.add_argument('--plan', help='Apply an already reviewed plan; omitted means read-only preflight')
|
||||
parser.add_argument('--plan-sha256', help='Required exact plan SHA256 when applying')
|
||||
args = parser.parse_args(argv)
|
||||
run(args.db,args.out,pg_service=args.pg_service,pg_schema=args.pg_schema,plan=args.plan,plan_sha256=args.plan_sha256)
|
||||
@@ -12,6 +12,69 @@ from paper_atlas.collectors.public_openreview import SOURCE as PUBLIC_SOURCE
|
||||
from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE
|
||||
|
||||
|
||||
def coverage_options(manifest):
|
||||
mode = manifest.get('coverage_mode', 'snapshot')
|
||||
if mode not in ('snapshot', 'incremental'):
|
||||
raise ValueError('Unsupported coverage mode')
|
||||
if mode == 'snapshot':
|
||||
return None
|
||||
required = ('venue', 'year', 'track', 'count', 'coverage_previous_count',
|
||||
'coverage_previous_expected_count', 'coverage_expected_count')
|
||||
if any(key not in manifest for key in required) or manifest.get('status') != 'partial':
|
||||
raise ValueError('Incremental coverage requires a partial, explicitly scoped manifest')
|
||||
previous = manifest['coverage_previous_count']
|
||||
expected = manifest['coverage_expected_count']
|
||||
prior_expected = manifest['coverage_previous_expected_count']
|
||||
if (type(previous) is not int or previous < 0 or type(expected) is not int or expected < 1
|
||||
or type(manifest['count']) is not int or manifest['count'] < 1
|
||||
or expected < max(previous, manifest['count'])
|
||||
or (prior_expected is not None and (type(prior_expected) is not int or prior_expected < 0
|
||||
or expected < prior_expected))):
|
||||
raise ValueError('Invalid incremental coverage counts')
|
||||
return previous, prior_expected, expected
|
||||
|
||||
|
||||
def source_record_count(c, source, edition):
|
||||
return c.execute('''SELECT count(DISTINCT ss.source_record_id) FROM submission_sources ss
|
||||
JOIN submissions s ON s.id=ss.submission_id
|
||||
JOIN source_records sr ON sr.id=ss.source_record_id
|
||||
WHERE s.edition_id=? AND sr.source_id=?''', (edition, source)).fetchone()[0]
|
||||
|
||||
|
||||
def require_incremental_parent(c, source, manifest, options):
|
||||
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 []
|
||||
previous, prior_expected, _ = options
|
||||
if entries:
|
||||
if len(entries) != 1 or entries[0][0] != manifest['list_url']:
|
||||
raise ValueError('Incremental coverage requires one unambiguous source listing')
|
||||
if tuple(entries[0][1:]) != ('partial', prior_expected, previous):
|
||||
raise ValueError('Incremental coverage parent changed or is not partial')
|
||||
elif previous != 0 or prior_expected is not None:
|
||||
raise ValueError('Incremental coverage parent is absent')
|
||||
actual = source_record_count(c, source, edition[0]) if edition else 0
|
||||
if actual != previous:
|
||||
raise ValueError('Incremental coverage parent differs from actual source records')
|
||||
|
||||
|
||||
def ingest_file(db,path, *, pg_service=None, pg_schema=None):
|
||||
path = Path(path)
|
||||
manifest_path = path.with_suffix('.manifest.json')
|
||||
@@ -24,6 +87,7 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
|
||||
source_name = manifest.get('source')
|
||||
if not source_name:
|
||||
raise ValueError('A source manifest is required')
|
||||
incremental = coverage_options(manifest)
|
||||
store = open_store(db, pg_service=pg_service, pg_schema=pg_schema)
|
||||
c = store.conn
|
||||
run = None
|
||||
@@ -35,6 +99,8 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
|
||||
source_id = store.source(source_name,manifest['list_url'])
|
||||
run = inserted_id(c,'INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path),'expected_sha256':manifest.get('file_sha256')})))
|
||||
c.commit()
|
||||
if incremental is not None:
|
||||
require_incremental_parent(c, source_id, manifest, incremental)
|
||||
digest = hashlib.sha256()
|
||||
with path.open('rb') as stream:
|
||||
for line in stream:
|
||||
@@ -42,9 +108,11 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
|
||||
row = json.loads(line)
|
||||
if row['source'] != source_name:
|
||||
raise ValueError('Source differs from manifest')
|
||||
if any(key in manifest and manifest[key]!=row[key] for key in ('venue','year','track')):
|
||||
if any(key in manifest and manifest[key]!=row.get(key) for key in ('venue','year','track','event_key')):
|
||||
raise ValueError('Conference edition differs from manifest')
|
||||
require_track_record(row, manifest)
|
||||
from paper_atlas.db.events import validate_event_key
|
||||
validate_event_key(row.get('event_key'))
|
||||
external = row['external_id']
|
||||
if external in seen:
|
||||
raise ValueError('Duplicate external ID in one collected file: '+external)
|
||||
@@ -65,7 +133,7 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
|
||||
status=event['status']
|
||||
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))
|
||||
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'))
|
||||
editions.add(edition)
|
||||
else:
|
||||
raise ValueError('This importer requires an explicit supported decision status')
|
||||
@@ -82,8 +150,25 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
|
||||
if expected is not None and expected != count:
|
||||
raise ValueError(f'Manifest count {expected} differs from records {count}')
|
||||
complete=expected==count and manifest.get('status') not in ('partial','blocked') and not manifest.get('missing_abstract_count',0) and not missing_abstracts
|
||||
c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) 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_success_at=excluded.last_success_at,last_error=NULL',(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,count,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())))
|
||||
c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps({'file':str(path),'file_sha256':digest.hexdigest()}),run))
|
||||
imported = count
|
||||
condition = ''
|
||||
condition_params = ()
|
||||
if incremental is not None:
|
||||
previous, prior_expected, expected = incremental
|
||||
imported = source_record_count(c, source_id, edition)
|
||||
if imported < previous or imported > expected:
|
||||
raise ValueError('Incremental coverage exceeds the declared source bounds')
|
||||
complete = False
|
||||
condition = " WHERE edition_sources.imported_count=? AND COALESCE(edition_sources.expected_count,-1)=? AND edition_sources.status='partial'"
|
||||
condition_params = (previous, -1 if prior_expected is None else prior_expected)
|
||||
result = c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) 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_success_at=excluded.last_success_at,last_error=NULL'+condition,(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,imported,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),*condition_params))
|
||||
if incremental is not None and result.rowcount != 1:
|
||||
raise ValueError('Incremental coverage parent changed during import')
|
||||
checkpoint = {'file':str(path),'file_sha256':digest.hexdigest()}
|
||||
if incremental is not None:
|
||||
checkpoint['coverage'] = {'mode':'incremental','previous_count':previous,
|
||||
'expected_count':expected,'imported_count':imported}
|
||||
c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps(checkpoint),run))
|
||||
c.commit()
|
||||
print(json.dumps({'file':str(path),'status':'imported','records':count,'missing_abstract_count':missing_abstracts}),flush=True)
|
||||
except BaseException as exc:
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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) + ')')]
|
||||
|
||||
@@ -85,3 +85,5 @@ class WriteConnection:
|
||||
class PostgresStore(Store):
|
||||
def __init__(self, service, schema=None):
|
||||
self.conn = WriteConnection(service, schema)
|
||||
from paper_atlas.db.events import has_event_key
|
||||
self.event_identity = has_event_key(self.conn)
|
||||
|
||||
+38
-4
@@ -14,7 +14,9 @@ ACCEPTED_PAPERS = "SELECT s.paper_id FROM submissions s JOIN decisions d ON d.su
|
||||
|
||||
|
||||
def known_acceptances(conn,paper):
|
||||
return [dict(row) for row in conn.execute('''SELECT s.id AS submission_id,v.acronym AS venue,e.year,e.track,
|
||||
from paper_atlas.db.events import has_event_key
|
||||
event = 'e.event_key' if has_event_key(conn) else "''"
|
||||
return [dict(row) for row in conn.execute('''SELECT s.id AS submission_id,v.acronym AS venue,e.year,e.track,'''+event+''' AS event_key,
|
||||
d.id AS decision_id,d.decision_date,d.observed_at,d.evidence_url
|
||||
FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
|
||||
JOIN decisions d ON d.submission_id=s.id WHERE s.paper_id=? AND d.status='accepted' AND '''+LATEST_DECISION+
|
||||
@@ -29,10 +31,13 @@ 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)
|
||||
parser.add_argument('--track',help='Exact track, e.g. main, findings, workshop')
|
||||
parser.add_argument('--event-key',help='Exact event within a venue/year/track')
|
||||
parser.add_argument('--status', choices=['accepted','rejected','desk_rejected','withdrawn','under_review','unknown'])
|
||||
parser.add_argument('--outcome',choices=['accepted','no_known_acceptance'],help='跨会议的已知录用情况;未查到录用不等于从未录用')
|
||||
parser.add_argument('--limit', type=int, default=20)
|
||||
@@ -47,6 +52,13 @@ 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.db.events import has_event_key
|
||||
event = 'e.event_key' if has_event_key(conn) else "''"
|
||||
if args.event_key is not None and event == "''":
|
||||
conn.close()
|
||||
raise ValueError('Event identity migration required for event-key queries')
|
||||
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,17 +81,35 @@ 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:
|
||||
if args.venue or args.year or args.status or args.track or args.event_key is not None:
|
||||
clause = ['1=1']
|
||||
if args.venue:
|
||||
if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='venue_aliases'").fetchone():
|
||||
clause.append('(lower(v.acronym)=lower(?) OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND lower(va.alias)=lower(?)))'); params.extend([args.venue,args.venue])
|
||||
else:
|
||||
clause.append('lower(v.acronym)=lower(?)');params.append(args.venue)
|
||||
if args.event_key is not None:
|
||||
clause.append('e.event_key=?');params.append(args.event_key)
|
||||
if args.track:
|
||||
clause.append('e.track=?');params.append(args.track)
|
||||
if args.year:
|
||||
@@ -98,10 +128,14 @@ 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 = []
|
||||
for sub in conn.execute('SELECT s.id,s.source_submission_id,s.submitted_at,v.acronym,e.year,e.track FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE s.paper_id=? ORDER BY e.year,s.id',(row['id'],)):
|
||||
for sub in conn.execute('SELECT s.id,s.source_submission_id,s.submitted_at,v.acronym,e.year,e.track,'+event+' AS event_key FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE s.paper_id=? ORDER BY e.year,s.id',(row['id'],)):
|
||||
submission = dict(sub)
|
||||
submission['decisions'] = []
|
||||
for d in conn.execute('SELECT id,status,raw_value,evidence_url,observed_at,decision_date FROM decisions WHERE submission_id=? ORDER BY observed_at,COALESCE(decision_date,observed_at),id',(sub['id'],)):
|
||||
|
||||
+1
-1
@@ -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"]
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
from contextlib import closing, redirect_stdout
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
import uuid
|
||||
|
||||
from paper_atlas import querying
|
||||
from paper_atlas.curation.annotations import (assert_annotation, available, initialize, main,
|
||||
observations, retract_annotation)
|
||||
from paper_atlas.db.backend import open_store
|
||||
from paper_atlas.db.store import Store
|
||||
from paper_atlas.postgres import schema
|
||||
from paper_atlas.postgres.client import connect
|
||||
from paper_atlas.postgres.migration import migrate
|
||||
|
||||
|
||||
class AnnotationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
|
||||
self.db = Path(self.tmp.name) / 'sample.sqlite'
|
||||
self.store = Store(self.db); self.addCleanup(self.store.conn.close)
|
||||
self.c = self.store.conn
|
||||
self.paper, self.record = self.store.record('fixture', 'one', 'https://example.org/one', 'First research', None, [])
|
||||
self.other, self.other_record = self.store.record('fixture', 'two', 'https://example.org/two', 'Other research', None, [])
|
||||
self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'original'}])
|
||||
self.c.commit()
|
||||
|
||||
def add(self, **changes):
|
||||
args = dict(paper=self.paper, kind='tag', namespace='custom:topic', value='reading',
|
||||
request_id='request-one', evidence_key='manual:reading', reason='Reviewed metadata', actor='reviewer')
|
||||
args.update(changes)
|
||||
return assert_annotation(self.c, **args)
|
||||
|
||||
def query(self, *args):
|
||||
self.c.commit()
|
||||
with redirect_stdout(io.StringIO()) as output:
|
||||
querying.main(['--db', str(self.db), *map(str, args)])
|
||||
return json.loads(output.getvalue())
|
||||
|
||||
def test_read_does_not_install_extension_and_write_requires_explicit_init(self):
|
||||
self.assertFalse(available(self.c))
|
||||
self.assertEqual(observations(self.c, self.paper), [])
|
||||
with self.assertRaisesRegex(RuntimeError, 'init explicitly'):
|
||||
self.add()
|
||||
self.assertEqual(self.query('--research-type', 'custom:method=empirical'), [])
|
||||
self.assertFalse(available(self.c))
|
||||
self.assertEqual(len(schema.inspect(self.c)), 21)
|
||||
|
||||
def test_schema_upgrade_is_transactional_and_marker_mismatch_fails_closed(self):
|
||||
self.assertTrue(initialize(self.c)); self.c.rollback()
|
||||
self.assertFalse(available(self.c))
|
||||
self.c.execute("INSERT INTO schema_migrations(name) VALUES('annotation_events_v1')")
|
||||
with self.assertRaisesRegex(RuntimeError, 'mismatch'):
|
||||
initialize(self.c)
|
||||
self.c.rollback()
|
||||
self.assertTrue(initialize(self.c)); self.assertFalse(initialize(self.c))
|
||||
self.assertEqual(len(schema.inspect(self.c)), 22)
|
||||
|
||||
def test_unmarked_table_cannot_receive_assertions(self):
|
||||
initialize(self.c)
|
||||
self.c.execute("DELETE FROM schema_migrations WHERE name='annotation_events_v1'")
|
||||
with self.assertRaisesRegex(RuntimeError, 'init explicitly'):
|
||||
self.add()
|
||||
self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0)
|
||||
|
||||
def test_types_tags_and_platform_labels_are_independent_and_queryable(self):
|
||||
initialize(self.c)
|
||||
self.add()
|
||||
self.add(kind='research_type', namespace='custom:topic', request_id='type', value='reading')
|
||||
self.assertEqual(len(self.query('--tag', 'custom:topic=reading')), 1)
|
||||
self.assertEqual(len(self.query('--research-type', 'custom:topic=reading')), 1)
|
||||
self.assertEqual(len(self.query('--tag', 'fixture:topic=original')), 1)
|
||||
detail = self.query('--paper', self.paper)[0]
|
||||
self.assertEqual(len(detail['annotation_tags']), 1)
|
||||
self.assertEqual(len(detail['research_types']), 1)
|
||||
self.assertNotIn('annotation_history', detail)
|
||||
self.assertEqual(detail['tags'][0]['namespace'], 'fixture:topic')
|
||||
|
||||
def test_repeated_request_is_idempotent_and_changed_content_fails(self):
|
||||
initialize(self.c); event = self.add()
|
||||
self.assertEqual(self.add(), event)
|
||||
for change in ({'value':'different'}, {'kind':'research_type'}, {'reason':'changed'}, {'paper':self.other}):
|
||||
with self.assertRaisesRegex(ValueError, 'different annotation'):
|
||||
self.add(**change)
|
||||
self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 1)
|
||||
|
||||
def test_retraction_keeps_original_and_other_assertions_of_same_label(self):
|
||||
initialize(self.c); first = self.add(); second = self.add(request_id='second', evidence_key='independent')
|
||||
kwargs = dict(request_id='retract', reason='No longer relevant', actor='reviewer')
|
||||
removed = retract_annotation(self.c, first, **kwargs)
|
||||
self.assertEqual(retract_annotation(self.c, first, **kwargs), removed)
|
||||
self.assertEqual([r['id'] for r in observations(self.c, self.paper)], [second])
|
||||
self.assertEqual(len(self.query('--paper', self.paper, '--annotation-history')[0]['annotation_history']), 3)
|
||||
with self.assertRaisesRegex(ValueError, 'already retracted'):
|
||||
retract_annotation(self.c, first, **{**kwargs,'request_id':'duplicate'})
|
||||
with self.assertRaisesRegex(ValueError, 'existing assertion'):
|
||||
retract_annotation(self.c, removed, **{**kwargs,'request_id':'nested'})
|
||||
|
||||
def test_source_assertions_require_matching_paper_and_platform(self):
|
||||
initialize(self.c)
|
||||
for change in ({'namespace':'fixture:type'}, {'namespace':'other:type','source_record_id':self.record},
|
||||
{'namespace':'fixture:type','source_record_id':self.other_record}):
|
||||
with self.assertRaises(ValueError):
|
||||
self.add(**change)
|
||||
event = self.add(kind='research_type', namespace='fixture:declared_type', source_record_id=self.record)
|
||||
with self.assertRaisesRegex(ValueError, 'Manual CLI'):
|
||||
retract_annotation(self.c, event, request_id='cannot', reason='x', actor='y', custom_only=True)
|
||||
self.assertEqual(observations(self.c, self.paper)[0]['source_record_id'], self.record)
|
||||
|
||||
def test_platform_tag_refresh_preserves_both_custom_kinds(self):
|
||||
initialize(self.c); self.add(); self.add(kind='research_type', request_id='type')
|
||||
self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'refreshed'}])
|
||||
self.assertEqual(len(observations(self.c, self.paper)), 2)
|
||||
self.assertEqual(len(self.query('--tag','fixture:topic=refreshed')), 1)
|
||||
self.assertEqual(self.query('--tag','fixture:topic=original'), [])
|
||||
|
||||
def test_merge_preserves_retractions_evidence_and_original_paper_ids(self):
|
||||
initialize(self.c); event = self.add(source_record_id=self.record)
|
||||
removed = retract_annotation(self.c, event, request_id='withdraw', reason='corrected', actor='reviewer')
|
||||
active = self.add(request_id='active', kind='research_type', value='survey')
|
||||
self.store.merge(self.other, self.paper, 'Independently reviewed identity')
|
||||
rows = observations(self.c, self.other, history=True)
|
||||
self.assertEqual([r['id'] for r in rows], [event, removed, active])
|
||||
self.assertEqual({r['original_paper_id'] for r in rows}, {self.paper})
|
||||
self.assertEqual({r['paper_id'] for r in rows}, {self.other})
|
||||
self.assertEqual([r['id'] for r in observations(self.c, self.other)], [active])
|
||||
self.assertEqual(self.add(source_record_id=self.record), event) # exact retry survives canonical merge
|
||||
self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(), [])
|
||||
|
||||
def test_invalid_input_does_not_create_events(self):
|
||||
initialize(self.c)
|
||||
for change in ({'paper':True}, {'paper':999999}, {'kind':'main'}, {'namespace':'arxiv'},
|
||||
{'namespace':'custom:topic;DROP'}, {'value':' '}, {'reason':''},
|
||||
{'actor':''}, {'request_id':''}, {'evidence_key':''}, {'source_record_id':True},
|
||||
{'evidence_url':'file:///etc/passwd'}, {'evidence_url':'https://user:secret@example.org/'}):
|
||||
with self.subTest(change=change), self.assertRaises(ValueError):
|
||||
self.add(**change)
|
||||
self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0)
|
||||
|
||||
def test_cli_manual_scope_and_complete_assert_retract_lifecycle(self):
|
||||
with redirect_stdout(io.StringIO()):
|
||||
main(['init','--db',str(self.db)])
|
||||
main(['assert','--db',str(self.db),'--paper',str(self.paper),'--kind','research_type',
|
||||
'--namespace','custom:method','--value','empirical','--evidence-key','manual:review',
|
||||
'--request-id','cli-one','--reason','Read abstract','--actor','reviewer'])
|
||||
self.assertEqual(len(self.query('--research-type','custom:method=empirical')), 1)
|
||||
event = observations(self.c, self.paper)[0]['id']
|
||||
with redirect_stdout(io.StringIO()):
|
||||
main(['retract','--db',str(self.db),'--event',str(event),'--request-id','cli-two',
|
||||
'--reason','Corrected classification','--actor','reviewer'])
|
||||
self.assertEqual(self.query('--research-type','custom:method=empirical'), [])
|
||||
with redirect_stdout(io.StringIO()) as out:
|
||||
main(['list','--db',str(self.db),'--paper',str(self.paper),'--history'])
|
||||
self.assertEqual(len(json.loads(out.getvalue())), 2)
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires approved remote PostgreSQL service')
|
||||
class PostgreSQLAnnotationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
|
||||
self.target = 'pa_anno_' + uuid.uuid4().hex[:12]
|
||||
self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
|
||||
self.db = Path(self.tmp.name) / 'source.sqlite'
|
||||
self.s = Store(self.db)
|
||||
self.paper, self.record = self.s.record('fixture','one','https://example.org','Paper',None,[])
|
||||
self.s.conn.commit()
|
||||
self.addCleanup(self.s.conn.close); self.addCleanup(self.cleanup)
|
||||
|
||||
def cleanup(self):
|
||||
with connect(self.service) as c:
|
||||
for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.target,)):
|
||||
c.execute('DROP SCHEMA ' + schema.quote(name) + ' CASCADE')
|
||||
|
||||
def migrate(self):
|
||||
self.s.conn.commit()
|
||||
self.s.conn.close() # Freeze/checkpoint the fixture, just like a real source migration.
|
||||
with redirect_stdout(io.StringIO()):
|
||||
return migrate(self.db, self.service, Path(self.tmp.name)/'report.json', schema=self.target, min_free_gib=0)
|
||||
|
||||
def args(self, **changes):
|
||||
a = dict(paper=self.paper,kind='research_type',namespace='custom:method',value='survey\x00and\x01',
|
||||
request_id='native',evidence_key='manual',reason='Reviewed\x00evidence',actor='reviewer')
|
||||
a.update(changes); return a
|
||||
|
||||
def test_native_schema_upgrade_roundtrip_retraction_and_queries(self):
|
||||
self.migrate()
|
||||
with closing(open_store(pg_service=self.service, pg_schema=self.target).conn) as c:
|
||||
self.assertTrue(initialize(c)); self.assertFalse(initialize(c))
|
||||
event = assert_annotation(c, **self.args()); c.commit()
|
||||
self.assertEqual(assert_annotation(c, **self.args()), event)
|
||||
self.assertEqual(observations(c,self.paper)[0]['value'], 'survey\x00and\x01')
|
||||
c.commit()
|
||||
with redirect_stdout(io.StringIO()) as out:
|
||||
querying.main(['--pg-service',self.service,'--pg-schema',self.target,'--paper',str(self.paper)])
|
||||
self.assertEqual(json.loads(out.getvalue())[0]['research_types'][0]['id'], event)
|
||||
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
|
||||
retract_annotation(c,event,request_id='removed',reason='Changed',actor='reviewer');c.commit()
|
||||
self.assertEqual(observations(c,self.paper), [])
|
||||
self.assertEqual(len(observations(c,self.paper,history=True)), 2)
|
||||
|
||||
def test_initialized_sqlite_extension_migrates_all_rows_hashes_and_identity(self):
|
||||
initialize(self.s.conn)
|
||||
event = assert_annotation(self.s.conn, **self.args())
|
||||
retract_annotation(self.s.conn,event,request_id='retracted',reason='Corrected',actor='reviewer')
|
||||
report = self.migrate()
|
||||
self.assertEqual(len(report['tables']),22)
|
||||
self.assertEqual(report['tables']['annotation_events']['source_rows'],2)
|
||||
self.assertEqual(report['tables']['annotation_events']['source_sha256'],report['tables']['annotation_events']['target_sha256'])
|
||||
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
|
||||
new = assert_annotation(c, **self.args(request_id='after-migration'))
|
||||
self.assertGreater(new,2); c.commit()
|
||||
self.assertEqual(len(observations(c,self.paper,history=True)),3)
|
||||
|
||||
def test_native_upgrade_rollback_keeps_legacy_schema_usable(self):
|
||||
self.migrate()
|
||||
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
|
||||
initialize(c); c.rollback()
|
||||
self.assertFalse(available(c))
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
|
||||
self.assertTrue(initialize(c))
|
||||
with self.assertRaises(ValueError):
|
||||
assert_annotation(c, **self.args(namespace='wrong:type',source_record_id=self.record))
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM annotation_events').fetchone()[0],0)
|
||||
c.rollback()
|
||||
@@ -7,6 +7,34 @@ from paper_atlas.audits.coverage import report
|
||||
from paper_atlas.db.store import Store
|
||||
|
||||
|
||||
def add_coverage_evidence(store):
|
||||
paper, record = store.record('official_metadata', 'known', 'https://example.org/known',
|
||||
'Known paper with two source observations', 'Complete canonical abstract', ['Jane Doe', 'John Smith'],
|
||||
{'openreview': 'coverageForum', 'arxiv': '2401.98765'})
|
||||
edition = store.decision(paper, record, 'COLING+LREC', 2024, 'main', 'known',
|
||||
'https://example.org/accept', 'accepted', 'Accept', aliases=['COLING', 'LREC'],
|
||||
identities=[('openreview', 'coverageForum')], observed_at='2025-05-01T00:00:00Z')
|
||||
same, mirror = store.record('bibliography', 'mirror', 'https://example.org/mirror',
|
||||
'Known paper with two source observations', None, ['Jane Doe', 'John Smith'], {'openreview': 'coverageForum'})
|
||||
store.decision(same, mirror, 'COLING+LREC', 2024, 'main', 'mirror', 'https://example.org/older',
|
||||
'rejected', 'Reject', identities=[('openreview', 'coverageForum')],
|
||||
observed_at='2025-05-01T00:00:00Z', decision_date='2025-04-30T00:00:00Z')
|
||||
unknown, unknown_record = store.record('unknown_catalogue', 'unknown', 'https://example.org/unknown',
|
||||
'Indexed work without a public decision', None, [])
|
||||
store.decision(unknown, unknown_record, 'COLING+LREC', 2024, 'main', 'unknown',
|
||||
'https://example.org/unknown', None, '', identities=[('openreview', 'unknownForum')])
|
||||
same, later = store.record('official_metadata', 'later', 'https://example.org/later',
|
||||
'Known paper with two source observations', 'Complete canonical abstract', ['Jane Doe', 'John Smith'],
|
||||
{'openreview': 'coverageNext', 'arxiv': '2401.98765'})
|
||||
store.decision(same, later, 'COLING+LREC', 2025, 'findings', 'later', 'https://example.org/later',
|
||||
'accepted', 'Accept', identities=[('openreview', 'coverageNext')], observed_at='2025-06-01T00:00:00Z')
|
||||
source = store.source('official_metadata', 'https://example.org')
|
||||
for url, status in [('https://example.org/list', 'complete'), ('https://example.org/identity-refresh', 'partial')]:
|
||||
store.conn.execute('''INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count)
|
||||
VALUES(?,?,?,?,?,?)''', (edition, source, url, status, 1, 1))
|
||||
store.conn.commit()
|
||||
|
||||
|
||||
class CoverageTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
|
||||
@@ -31,5 +59,46 @@ class CoverageTests(unittest.TestCase):
|
||||
self.store.conn.commit();coling=report(self.db,self.registry)['venues'][0]
|
||||
self.assertEqual(coling['imported_submissions'],2);self.assertEqual(coling['distinct_papers'],1)
|
||||
|
||||
def test_observations_attempts_and_missing_source_abstracts_stay_separate(self):
|
||||
add_coverage_evidence(self.store)
|
||||
result = report(self.db, self.registry)
|
||||
self.assertEqual(result['engine'], 'sqlite')
|
||||
self.assertEqual(result['database'], str(self.db))
|
||||
coling = result['venues'][0]
|
||||
self.assertEqual((coling['imported_submissions'], coling['distinct_papers']), (3, 2))
|
||||
edition, findings = coling['editions']
|
||||
self.assertEqual((edition['submissions'], edition['papers'], edition['papers_without_abstract_text']), (2, 2, 1))
|
||||
self.assertEqual(edition['latest_observed_status_counts']['accepted'], 1)
|
||||
self.assertEqual(edition['latest_observed_status_counts']['rejected'], 0)
|
||||
self.assertEqual(edition['latest_observed_status_counts']['unknown'], 1)
|
||||
self.assertEqual(edition['decision_observation_counts']['accepted'], 1)
|
||||
self.assertEqual(edition['decision_observation_counts']['rejected'], 1)
|
||||
self.assertEqual(edition['decision_observation_counts']['unknown'], 0)
|
||||
counts = {row['source']: row for row in edition['source_record_counts']}
|
||||
self.assertEqual(counts['bibliography']['source_records_without_abstract_text'], 1)
|
||||
self.assertEqual(counts['official_metadata']['source_records'], 1)
|
||||
self.assertEqual(len(edition['sources']), 2)
|
||||
self.assertEqual(findings['track'], 'findings')
|
||||
self.assertEqual(result['imported_venues_outside_registry'], [])
|
||||
|
||||
def test_missing_sqlite_audit_does_not_create_a_database(self):
|
||||
path = self.root / 'absent.sqlite'
|
||||
with self.assertRaisesRegex(ValueError, 'Existing database'):
|
||||
report(path, self.registry)
|
||||
self.assertFalse(path.exists())
|
||||
|
||||
def test_empty_editions_remain_visible_after_shared_aggregation(self):
|
||||
self.store.conn.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES('AAAI','AAAI','general_ai','conference')")
|
||||
venue = self.store.conn.execute("SELECT id FROM venues WHERE acronym='AAAI'").fetchone()[0]
|
||||
self.store.conn.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2026,'main')", (venue,))
|
||||
self.store.conn.commit()
|
||||
aaai = report(self.db, self.registry)['venues'][1]
|
||||
self.assertEqual(aaai['history_status'], 'no_imported_records')
|
||||
self.assertEqual(len(aaai['editions']), 1)
|
||||
edition = aaai['editions'][0]
|
||||
self.assertEqual((edition['submissions'], edition['papers'], edition['papers_without_abstract_text']), (0, 0, 0))
|
||||
self.assertEqual(sum(edition['latest_observed_status_counts'].values()), 0)
|
||||
self.assertEqual(edition['source_record_counts'], [])
|
||||
|
||||
|
||||
if __name__=='__main__':unittest.main()
|
||||
|
||||
@@ -10,7 +10,7 @@ from unittest.mock import patch
|
||||
|
||||
from paper_atlas.collectors.dblp import XML_SHA
|
||||
from paper_atlas.collectors.dblp_records import collect,identifiers,normalize_row,material_reason
|
||||
from paper_atlas.collectors.dblp_scope import scope
|
||||
from paper_atlas.collectors.dblp_scope import scope, HISTORICAL_WORKSHOP_TITLES
|
||||
from paper_atlas.db.store import Store
|
||||
from paper_atlas.ingestion.records import ingest_file
|
||||
|
||||
@@ -70,6 +70,69 @@ class DblpRecordTests(unittest.TestCase):
|
||||
self.assertEqual(scope(self.row(),self.volume('kdd',2024))[1],'crossref_venue_conflict')
|
||||
self.assertEqual(scope(self.row(year=2027),self.volume('aaai',2027))[1],'future_edition')
|
||||
|
||||
def test_historical_primary_series_workshop_names_require_exact_editions(self):
|
||||
self.assertEqual(set(HISTORICAL_WORKSHOP_TITLES),
|
||||
{('aistats',year) for year in [1995,1997,1999,2001,2003,2005]} |
|
||||
{('colt',year) for year in [1988,1989,1990,1991]})
|
||||
for (family,year),prefix in HISTORICAL_WORKSHOP_TITLES.items():
|
||||
with self.subTest(family=family,year=year):
|
||||
title=prefix+f', {family.upper()} {year}, Proceedings.'
|
||||
volume=self.volume(family,year,title)
|
||||
row=self.row(family,family.upper(),year)
|
||||
if year==1995:
|
||||
row['fields']['year']=['1996'];volume['fields']['year']=['1996']
|
||||
edition,reason=scope(row,volume)
|
||||
self.assertIsNone(reason);self.assertEqual(edition['track'],'main')
|
||||
self.assertEqual(edition['year'],year)
|
||||
self.assertIn('Historical primary series',edition['track_scope_note'])
|
||||
record,reason=normalize_row(row,{volume['key']:volume},{family.upper()})
|
||||
self.assertIsNone(reason);self.assertEqual(record['metadata']['track_scope_evidence_url'],edition['track_scope_evidence_url'])
|
||||
self.assertEqual(record['metadata']['proceedings_title'],title)
|
||||
if year==1995:self.assertEqual(record['metadata']['bibliographic_year'],1996)
|
||||
|
||||
def test_historical_exception_does_not_promote_affiliated_or_wrong_scope_workshops(self):
|
||||
title=HISTORICAL_WORKSHOP_TITLES[('aistats',1995)]+', AISTATS 1995, Proceedings.'
|
||||
for book,key,changed in [
|
||||
('AISTATS','conf/aistats/1995w',title),
|
||||
('AISTATS (Workshops)','conf/aistats/1995',title),
|
||||
('AISTATS','conf/aistats/1995',title.replace('Fifth','Sixth')),
|
||||
('AISTATS','conf/aistats/1995',title.replace('1995','1997'))]:
|
||||
volume=self.volume('aistats',1995,changed);volume['key']=key
|
||||
row=self.row('aistats',book,1995);row['fields']['crossref']=[key]
|
||||
self.assertEqual(scope(row,volume)[0]['track'],'workshop')
|
||||
volume=self.volume('aistats',2010,'Active Learning and Experimental Design workshop, In conjunction with AISTATS 2010')
|
||||
volume['key']='conf/aistats/2010w'
|
||||
row=self.row('aistats','AISTATS',2011);row['fields']['crossref']=[volume['key']]
|
||||
self.assertEqual(scope(row,volume)[0]['track'],'workshop')
|
||||
|
||||
def test_combined_colt_kernel_volume_keeps_unknown_per_paper_track(self):
|
||||
title='Computational Learning Theory and Kernel Machines, 16th Annual Conference on Computational Learning Theory and 7th Kernel Workshop, COLT/Kernel 2003, Washington, DC, USA, August 24-27, 2003, Proceedings'
|
||||
volume=self.volume('colt',2003,title);row=self.row('colt','COLT',2003)
|
||||
record,reason=normalize_row(row,{volume['key']:volume},{'COLT'})
|
||||
self.assertIsNone(reason);self.assertEqual(record['track'],'proceedings_unspecified')
|
||||
self.assertEqual(record['status'],'accepted')
|
||||
self.assertIn('per-paper section absent',record['metadata']['track_scope_note'])
|
||||
self.assertEqual(record['metadata']['track_scope_evidence_url'],'https://doi.org/10.1007/b12006')
|
||||
volume['key']='conf/colt/2003w';row['fields']['crossref']=[volume['key']]
|
||||
self.assertEqual(scope(row,volume)[0]['track'],'workshop')
|
||||
|
||||
def test_historical_scope_provenance_survives_file_ingestion(self):
|
||||
title=HISTORICAL_WORKSHOP_TITLES[('colt',1988)]+', COLT 1988, Proceedings.'
|
||||
row=self.row('colt','COLT',1988);volume=self.volume('colt',1988,title)
|
||||
record,reason=normalize_row(row,{volume['key']:volume},{'COLT'})
|
||||
self.assertIsNone(reason)
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root=Path(tmp);path=root/'history.jsonl'
|
||||
path.write_text(json.dumps(record)+'\n')
|
||||
path.with_suffix('.manifest.json').write_text(json.dumps({'source':'dblp','count':1,'list_url':'https://doi.org/10.4230/dblp.xml.2026-11-01','file_sha256':hashlib.sha256(path.read_bytes()).hexdigest()}))
|
||||
ingest_file(root/'sample.sqlite',path)
|
||||
with closing(sqlite3.connect(root/'sample.sqlite')) as connection:
|
||||
track=connection.execute('SELECT track FROM editions').fetchone()[0]
|
||||
metadata=json.loads(connection.execute('SELECT metadata_json FROM source_records').fetchone()[0])
|
||||
self.assertEqual(track,'main')
|
||||
self.assertEqual(metadata['track_scope_evidence_url'],'https://www.learningtheory.org/')
|
||||
self.assertEqual(metadata['proceedings_title'],title)
|
||||
|
||||
def test_only_own_identifier_links_are_used_and_ambiguity_is_preserved(self):
|
||||
own,ambiguous=identifiers({'ee':['https://doi.org/10.1234/published','https://doi.org/10.48550/arXiv.2401.00001','https://openreview.net/forum?id=realForum'],
|
||||
'cite':['https://doi.org/10.9999/reference']})
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
from contextlib import closing, redirect_stdout, contextmanager
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
import uuid
|
||||
from importlib.resources import files
|
||||
|
||||
from paper_atlas.db.store import Store
|
||||
from paper_atlas.db.backend import open_store
|
||||
from paper_atlas.db.events import has_event_key
|
||||
from paper_atlas.collectors.dblp import XML_SHA
|
||||
from paper_atlas.ingestion.event_identity import plan, upgrade, exact_counts
|
||||
from paper_atlas.ingestion.records import ingest_file
|
||||
from paper_atlas.ingestion.backfill import backfill
|
||||
from paper_atlas.postgres.client import connect
|
||||
from paper_atlas.postgres.schema import quote
|
||||
from paper_atlas.postgres.migration import migrate
|
||||
from paper_atlas import querying
|
||||
|
||||
|
||||
@contextmanager
|
||||
def opened_store(**kwargs):
|
||||
store=open_store(**kwargs)
|
||||
try:
|
||||
yield store
|
||||
finally:
|
||||
store.conn.close()
|
||||
|
||||
|
||||
def legacy(db):
|
||||
sql = files('paper_atlas.db').joinpath('schema.sql').read_text().replace(" event_key TEXT NOT NULL DEFAULT '',\n", '').replace('UNIQUE(venue_id,year,track,event_key)', 'UNIQUE(venue_id,year,track)')
|
||||
c = sqlite3.connect(db); c.executescript(sql); c.close()
|
||||
s = Store(db)
|
||||
p,r = s.record('dblp','conf/www/Fall95','https://example.org/fall','Same paper in two events',
|
||||
'Abstract\x00with\x01text',['A Person'],{'arxiv':'2501.00001'})
|
||||
s.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'WWW (Fall)'}])
|
||||
s.decision(p,r,'Legacy',1995,'main','conf/www/Fall95','https://example.org/accept','accepted','Accept')
|
||||
s.conn.execute("UPDATE venues SET acronym='WWW',name='WWW'")
|
||||
meta={'snapshot_sha256':XML_SHA,'proceedings_key':'conf/www/1995f','bibliographic_year':1995,
|
||||
'bibliography':{'booktitle':['WWW (Fall)'],'url':['db/conf/www/www1995f.html#Fall95']}}
|
||||
s.conn.execute('UPDATE source_records SET metadata_json=?',(json.dumps(meta),));s.conn.commit();s.conn.close()
|
||||
return p,r
|
||||
|
||||
|
||||
def add(store, external, event=None, status='accepted', venue='WWW', year=1995):
|
||||
p,r = store.record('fixture',external,'https://example.org/'+external,'Same paper in two events',
|
||||
'Abstract\x00with\x01text',['A Person'],{'arxiv':'2501.00001'})
|
||||
edition = store.decision(p,r,venue,year,'main',external,'https://example.org/'+status,status,status,event_key=event)
|
||||
return p,r,edition
|
||||
|
||||
|
||||
def file(root, name, rows, event=None):
|
||||
p=root/(name+'.jsonl');p.write_text(''.join(json.dumps(r)+'\n'for r in rows))
|
||||
manifest={'source':'fixture','venue':'WWW','year':1995,'track':'main','count':len(rows),'list_url':'https://example.org/'+name}
|
||||
if event is not None:manifest['event_key']=event
|
||||
p.with_suffix('.manifest.json').write_text(json.dumps(manifest));return p
|
||||
|
||||
|
||||
class EventIdentityTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.db=self.root/'sample.sqlite'
|
||||
|
||||
def test_same_paper_distinct_events_keep_both_attempts_and_decisions(self):
|
||||
s=Store(self.db);self.addCleanup(s.conn.close)
|
||||
paper,_,spring=add(s,'spring','www-1995-spring','rejected');_,_,fall=add(s,'fall','www-1995-fall')
|
||||
s.conn.commit();self.assertNotEqual(spring,fall)
|
||||
with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--paper',str(paper)])
|
||||
row=json.loads(out.getvalue())[0];self.assertEqual({a['event_key']for a in row['submissions']},{'www-1995-spring','www-1995-fall'})
|
||||
self.assertEqual([a['event_key']for a in row['known_acceptances']],['www-1995-fall'])
|
||||
self.assertEqual({a['latest_status']for a in row['submissions']},{'rejected','accepted'})
|
||||
with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--venue','WWW','--event-key','www-1995-spring','--status','rejected'])
|
||||
self.assertEqual(len(json.loads(out.getvalue())),1)
|
||||
|
||||
def test_explicit_migration_preserves_ids_history_and_source_bytes(self):
|
||||
paper,record=legacy(self.db);proposal=plan(self.db);self.assertFalse(proposal['before']['event_column'])
|
||||
result=upgrade(proposal,self.db);self.assertEqual(result['updated_editions'],1)
|
||||
self.assertEqual(result['after']['protected'],proposal['before']['protected'])
|
||||
self.assertEqual(result['after']['editions_sha256'],proposal['before']['editions_sha256'])
|
||||
self.assertEqual(upgrade(proposal,self.db)['status'],'already_applied')
|
||||
s=Store(self.db);self.addCleanup(s.conn.close)
|
||||
self.assertEqual(s.conn.execute('SELECT id,event_key FROM editions').fetchall(),[(1,'www-1995-fall')])
|
||||
# An old file without event_key still has its original source anchor.
|
||||
self.assertEqual(s.decision(paper,record,'WWW',1995,'main','conf/www/Fall95','https://example.org/accept','accepted','Accept'),1)
|
||||
self.assertEqual(s.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
|
||||
self.assertEqual(s.conn.execute('PRAGMA foreign_key_check').fetchall(),[])
|
||||
|
||||
def test_unverified_legacy_volume_is_not_automatically_labelled(self):
|
||||
legacy(self.db)
|
||||
with sqlite3.connect(self.db)as c:c.execute("UPDATE source_records SET metadata_json='{}'")
|
||||
with self.assertRaisesRegex(ValueError,'complete pinned Fall'):plan(self.db)
|
||||
with sqlite3.connect(self.db)as c:self.assertFalse(has_event_key(c))
|
||||
|
||||
def test_stale_plan_rolls_back_before_schema_change(self):
|
||||
legacy(self.db);proposal=plan(self.db)
|
||||
with sqlite3.connect(self.db)as c:c.execute("UPDATE editions SET track='unspecified'")
|
||||
with self.assertRaisesRegex(ValueError,'stale'):upgrade(proposal,self.db)
|
||||
with sqlite3.connect(self.db)as c:self.assertFalse(has_event_key(c));self.assertEqual(c.execute('SELECT track FROM editions').fetchone()[0],'unspecified')
|
||||
|
||||
def test_unscoped_source_and_wrong_refresh_event_fail(self):
|
||||
s=Store(self.db);self.addCleanup(s.conn.close);_,r,e=add(s,'first','www-1995-fall');s.conn.commit()
|
||||
p,r2=s.record('fixture','unscoped','https://example.org/u','Other work',None,[])
|
||||
with self.assertRaisesRegex(ValueError,'event_key'):s.decision(p,r2,'WWW',1995,'main','unscoped','https://example.org/u','accepted','Accept')
|
||||
with self.assertRaisesRegex(ValueError,'Cannot move'):s.decision(p,r,'WWW',1995,'main','first','https://example.org/u','accepted','Accept',event_key='www-1995-spring')
|
||||
self.assertEqual(s.conn.execute('SELECT count(*) FROM editions').fetchone()[0],1)
|
||||
|
||||
def test_pinned_fall_cannot_claim_spring(self):
|
||||
p,r=legacy(self.db);upgrade(plan(self.db),self.db);s=Store(self.db);self.addCleanup(s.conn.close)
|
||||
with self.assertRaisesRegex(ValueError,'pinned WWW Fall'):s.decision(p,r,'WWW',1995,'main','conf/www/Fall95','https://example.org/u','accepted','Accept',event_key='www-1995-spring')
|
||||
|
||||
def test_manifest_event_mismatch_rolls_back_the_whole_file(self):
|
||||
s=Store(self.db);s.conn.close()
|
||||
base={'source':'fixture','venue':'WWW','year':1995,'track':'main','status':'accepted','title':'Input work','url':'https://example.org/p','authors':[]}
|
||||
path=file(self.root,'bad',[dict(base,external_id='a',event_key='www-1995-spring'),dict(base,external_id='b',event_key='www-1995-fall')],'www-1995-spring')
|
||||
with self.assertRaisesRegex(ValueError,'edition differs'):ingest_file(self.db,path)
|
||||
with sqlite3.connect(self.db)as c:
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],0)
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM editions').fetchone()[0],0)
|
||||
|
||||
def test_backfill_selects_the_explicit_event_and_preserves_attempts(self):
|
||||
s=Store(self.db);add(s,'spring','www-1995-spring','rejected');add(s,'fall','www-1995-fall');s.conn.commit();s.conn.close()
|
||||
row={'source':'fixture','external_id':'fall','venue':'WWW','year':1995,'track':'main','event_key':'www-1995-fall','status':'accepted','identifiers':{'openreview':'forumFall'}}
|
||||
path=file(self.root,'backfill',[row],'www-1995-fall')
|
||||
with redirect_stdout(io.StringIO()):result=backfill(self.db,[path])
|
||||
self.assertEqual(result['submissions_after'],2)
|
||||
with sqlite3.connect(self.db)as c:
|
||||
self.assertEqual(c.execute("SELECT e.event_key FROM submission_identifiers si JOIN submissions s ON s.id=si.submission_id JOIN editions e ON e.id=s.edition_id WHERE si.scheme='openreview' AND si.value='forumFall'").fetchone()[0],'www-1995-fall')
|
||||
|
||||
def test_exact_paged_counts_cover_composite_key_boundaries(self):
|
||||
s=Store(self.db);self.addCleanup(s.conn.close)
|
||||
for i in range(7):add(s,'entry'+str(i),'www-1995-fall')
|
||||
s.conn.commit();events=[];counts=exact_counts(s.conn,page_size=2,progress=events.append)
|
||||
self.assertEqual(counts['source_records'],7)
|
||||
self.assertEqual(counts['identifiers'],8)
|
||||
self.assertEqual(counts['submission_sources'],7)
|
||||
self.assertEqual([e['rows_counted']for e in events if e['table']=='identifiers'and e['complete']],[8])
|
||||
|
||||
def test_legacy_single_event_writes_and_reads_remain_available(self):
|
||||
legacy(self.db);s=Store(self.db);self.addCleanup(s.conn.close)
|
||||
add(s,'ordinary',venue='COLM',year=2025);s.conn.commit()
|
||||
with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--venue','COLM'])
|
||||
self.assertEqual(json.loads(out.getvalue())[0]['known_acceptances'][-1]['event_key'],'')
|
||||
with self.assertRaisesRegex(ValueError,'migration required'):
|
||||
querying.main(['--db',str(self.db),'--event-key','www-1995-fall'])
|
||||
|
||||
def test_unexpected_extra_edition_column_requires_review(self):
|
||||
legacy(self.db)
|
||||
with sqlite3.connect(self.db)as c:c.execute('ALTER TABLE editions ADD COLUMN custom_data TEXT')
|
||||
with self.assertRaisesRegex(ValueError,'Unexpected edition schema'):plan(self.db)
|
||||
|
||||
def test_file_spanning_two_events_without_manifest_key_rolls_back(self):
|
||||
Store(self.db).conn.close()
|
||||
base={'source':'fixture','venue':'WWW','year':1995,'track':'main','status':'accepted','title':'Input work','url':'https://example.org/p','authors':[]}
|
||||
path=file(self.root,'two',[dict(base,external_id='a',event_key='www-1995-spring'),dict(base,external_id='b',event_key='www-1995-fall')])
|
||||
with self.assertRaisesRegex(ValueError,'exactly one'):ingest_file(self.db,path)
|
||||
with sqlite3.connect(self.db)as c:self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],0)
|
||||
|
||||
def test_cutover_sqlite_backup_remains_protected(self):
|
||||
legacy(self.db);proposal=plan(self.db);Path(str(self.db.resolve())+'.postgresql-cutover.json').write_text('{}')
|
||||
with self.assertRaisesRegex(RuntimeError,'cut over'):upgrade(proposal,self.db)
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL test service')
|
||||
class PostgreSQLEventIdentityTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_event_'+uuid.uuid4().hex[:12]
|
||||
t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.db=self.root/'seed.sqlite';legacy(self.db)
|
||||
self.addCleanup(self.cleanup)
|
||||
with redirect_stdout(io.StringIO()):migrate(self.db,self.service,self.root/'migrate.json',schema=self.schema,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.schema,)).fetchall():c.execute('DROP SCHEMA '+quote(name)+' CASCADE')
|
||||
|
||||
def test_native_upgrade_preserves_ids_and_codec_then_accepts_separate_event(self):
|
||||
proposal=plan(pg_service=self.service,pg_schema=self.schema)
|
||||
result=upgrade(proposal,pg_service=self.service,pg_schema=self.schema)
|
||||
self.assertEqual(result['after']['protected'],proposal['before']['protected'])
|
||||
self.assertEqual(upgrade(proposal,pg_service=self.service,pg_schema=self.schema)['status'],'already_applied')
|
||||
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:
|
||||
paper,_,spring=add(s,'spring','www-1995-spring','rejected');s.conn.commit()
|
||||
self.assertNotEqual(spring,1)
|
||||
self.assertEqual(s.conn.execute('SELECT abstract FROM papers WHERE id=?',(paper,)).fetchone()[0],'Abstract\x00with\x01text')
|
||||
with redirect_stdout(io.StringIO())as out:querying.main(['--pg-service',self.service,'--pg-schema',self.schema,'--paper',str(paper)])
|
||||
row=json.loads(out.getvalue())[0];self.assertEqual(len(row['submissions']),2);self.assertEqual(len(row['known_acceptances']),1)
|
||||
|
||||
def test_native_paged_counts_preserve_encoded_composite_key_order(self):
|
||||
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:
|
||||
for value in ['a\x00b','a\x01b','a\x02b','z']:
|
||||
s.conn.execute('INSERT INTO identifiers(scheme,value,paper_id) VALUES(?,?,?)',('custom',value,1))
|
||||
s.conn.commit();counts=exact_counts(s.conn,page_size=2)
|
||||
self.assertEqual(counts['identifiers'],6)
|
||||
self.assertEqual(counts['submissions'],1)
|
||||
|
||||
def test_stale_native_plan_is_rejected_and_lock_released(self):
|
||||
proposal=plan(pg_service=self.service,pg_schema=self.schema)
|
||||
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:s.conn.execute("UPDATE editions SET track='unspecified'");s.conn.commit()
|
||||
with self.assertRaisesRegex(ValueError,'stale'):upgrade(proposal,pg_service=self.service,pg_schema=self.schema)
|
||||
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:self.assertFalse(has_event_key(s.conn))
|
||||
@@ -0,0 +1,114 @@
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from paper_atlas.audits.histories import report
|
||||
from paper_atlas.db.store import Store
|
||||
|
||||
|
||||
def add_history_fixture(store):
|
||||
def event(work, forum, venue, year, track, status, observed):
|
||||
paper, record = store.record('history_fixture', work + '/' + forum,
|
||||
'https://example.org/' + forum, work + ' title\x00 preserved', 'Abstract', ['Jane Doe'],
|
||||
{'arxiv': work})
|
||||
store.decision(paper, record, venue, year, track, forum,
|
||||
'https://example.org/' + forum + '/' + str(status), status, status or '',
|
||||
identities=[('openreview', forum)], observed_at=observed)
|
||||
return paper
|
||||
|
||||
# The negative observation was collected later: observation order is not chronology.
|
||||
event('2501.00001', 'HistoryA01', 'ICLR', 2024, 'main', 'rejected', '2026-02-01T00:00:00Z')
|
||||
event('2501.00001', 'HistoryA02', 'ICML', 2025, 'main', 'accepted', '2026-01-01T00:00:00Z')
|
||||
event('2501.00002', 'HistoryB01', 'ICLR', 2024, 'main', 'rejected', '2026-01-01T00:00:00Z')
|
||||
event('2501.00002', 'HistoryB01', 'ICLR', 2024, 'main', 'accepted', '2026-02-01T00:00:00Z')
|
||||
event('2501.00003', 'HistoryC01', 'ICLR', 2024, 'main', 'desk_rejected', '2026-01-01T00:00:00Z')
|
||||
event('2501.00003', 'HistoryC02', 'ICLR', 2024, 'workshop_example', 'accepted', '2026-02-01T00:00:00Z')
|
||||
event('2501.00004', 'HistoryD01', 'ICML', 2025, 'main', 'accepted', '2026-01-01T00:00:00Z')
|
||||
event('2501.00004', 'HistoryD01', 'ICML', 2025, 'main', 'withdrawn', '2026-02-01T00:00:00Z')
|
||||
event('2501.00004', 'HistoryD02', 'ICLR', 2024, 'main', 'rejected', '2026-01-01T00:00:00Z')
|
||||
event('2501.00005', 'HistoryE01', 'COLM', 2026, 'main', None, '2026-01-01T00:00:00Z')
|
||||
store.conn.commit()
|
||||
|
||||
|
||||
class HistoryTests(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)
|
||||
|
||||
def test_distinct_attempts_venues_and_latest_acceptances_are_not_observation_counts(self):
|
||||
add_history_fixture(self.store)
|
||||
result = report(self.db)
|
||||
self.assertEqual(result['counts'], {
|
||||
'papers_with_submissions': 5, 'papers_with_multiple_submissions': 3,
|
||||
'papers_with_historical_negative': 4, 'papers_with_current_known_acceptance': 3,
|
||||
'papers_with_negative_and_acceptance_in_distinct_attempts': 2,
|
||||
'papers_with_negative_and_acceptance_in_distinct_venues': 1,
|
||||
'submissions': 8, 'decision_observations': 9,
|
||||
'attempts_with_historical_negative': 5,
|
||||
'attempts_with_historical_negative_and_acceptance': 2,
|
||||
})
|
||||
self.assertEqual(result['latest_attempt_status_counts'],
|
||||
{'accepted': 3, 'desk_rejected': 1, 'rejected': 2, 'unknown': 1, 'withdrawn': 1})
|
||||
self.assertEqual(len(result['examples']), 2)
|
||||
first = result['examples'][0]
|
||||
self.assertIn('\x00', first['title'])
|
||||
self.assertEqual({r['venue'] for r in first['attempts']}, {'ICLR', 'ICML'})
|
||||
self.assertTrue(all(r['decisions'][0]['sources'] for r in first['attempts']))
|
||||
self.assertTrue(all(r['decisions'][0]['decision_date'] is None for r in first['attempts']))
|
||||
self.assertIn('Not inferred', first['chronology'])
|
||||
self.assertEqual(result['status'], 'observed_history_audit_not_final_dedup')
|
||||
self.assertTrue(all(v == 0 for v in result['evidence_relationship_checks'].values()))
|
||||
|
||||
def test_missing_and_mislinked_source_evidence_is_reported_without_repair(self):
|
||||
add_history_fixture(self.store)
|
||||
self.store.conn.execute('DELETE FROM decision_sources WHERE decision_id=1')
|
||||
self.store.conn.execute('DELETE FROM submission_sources WHERE submission_id=1')
|
||||
# Record3 belongs to paper2; move it to paper1 to create a real mismatch.
|
||||
self.store.conn.execute('UPDATE source_records SET paper_id=1 WHERE id=3')
|
||||
self.store.conn.commit()
|
||||
result = report(self.db, examples=0)
|
||||
checks = result['evidence_relationship_checks']
|
||||
self.assertEqual(checks['decisions_without_source_links'], 1)
|
||||
self.assertEqual(checks['submissions_without_source_links'], 1)
|
||||
self.assertEqual(checks['submission_source_paper_mismatches'], 1)
|
||||
self.assertEqual(checks['decision_source_paper_mismatches'], 2)
|
||||
self.assertEqual(self.store.conn.execute('SELECT paper_id FROM source_records WHERE id=3').fetchone()[0], 1)
|
||||
|
||||
def test_example_limit_does_not_change_global_counts(self):
|
||||
add_history_fixture(self.store)
|
||||
none, one = report(self.db, examples=0), report(self.db, examples=1)
|
||||
self.assertEqual(none['counts'], one['counts'])
|
||||
self.assertEqual(none['examples'], [])
|
||||
self.assertEqual(len(one['examples']), 1)
|
||||
|
||||
def test_empty_snapshot_and_invalid_limit_do_not_create_missing_database(self):
|
||||
empty = report(self.db)
|
||||
self.assertTrue(all(v == 0 for v in empty['counts'].values()))
|
||||
self.assertEqual(empty['latest_attempt_status_counts'], {})
|
||||
absent = self.db.parent / 'absent.sqlite'
|
||||
with self.assertRaisesRegex(ValueError, 'Examples'):
|
||||
report(absent, examples=-1)
|
||||
with self.assertRaisesRegex(ValueError, 'Existing database'):
|
||||
report(absent)
|
||||
self.assertFalse(absent.exists())
|
||||
|
||||
def test_truncated_example_keeps_the_cross_venue_acceptance_visible(self):
|
||||
paper, record = self.store.record('fixture', 'many', 'https://example.org/many',
|
||||
'Many attempts', 'Abstract', [])
|
||||
for index in range(26):
|
||||
accepted = index == 25
|
||||
forum = f'BoundedForum{index}'
|
||||
self.store.decision(paper, record, 'ICML' if accepted else 'ICLR', 2025, 'main', forum,
|
||||
'https://example.org/' + forum, 'accepted' if accepted else 'rejected',
|
||||
'Accept' if accepted else 'Reject', identities=[('openreview', forum)])
|
||||
self.store.conn.commit()
|
||||
result = report(self.db)
|
||||
example, = result['examples']
|
||||
self.assertEqual((example['attempt_count'], len(example['attempts'])), (26, 20))
|
||||
self.assertTrue(example['attempts_truncated'])
|
||||
self.assertEqual({row['latest_observed_status'] for row in example['attempts']},
|
||||
{'accepted', 'rejected'})
|
||||
self.assertEqual(result['counts']['submissions'], 26)
|
||||
@@ -0,0 +1,185 @@
|
||||
import copy
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from paper_atlas.curation.identifiers import SCOPE, digest, json_bytes, prepare
|
||||
from paper_atlas.db.store import Store
|
||||
from paper_atlas.ingestion.records import ingest_file
|
||||
|
||||
|
||||
class IdentifierCurationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.output = self.root / 'prepared'
|
||||
self.row = {'source':'dblp','external_id':'conf/uai/example','url':'https://example.org/paper',
|
||||
'title':'A sufficiently long example research paper title','abstract':None,
|
||||
'authors':['Jane Doe','John Smith'],'venue':'UAI','year':2001,'track':'main',
|
||||
'status':'accepted','tags':[{'namespace':'dblp:topic','value':'version\u2029study\x00'}],
|
||||
'metadata':{'pages':'203-210','raw_title':'Original\u2029title'}}
|
||||
self.evidence = self.root / 'evidence.json'
|
||||
self.evidence.write_bytes(json_bytes({'url':'https://example.org/archive','comment':'Conference version'}))
|
||||
self.path = self.root / 'records.jsonl'
|
||||
self.raw = json_bytes(self.row)
|
||||
self.path.write_bytes(self.raw)
|
||||
self.manifest = {'source':'dblp','venue':'UAI','year':2001,'track':'main','count':1,
|
||||
'list_url':'https://example.org/list','file_sha256':digest(self.raw),
|
||||
'status':'partial','missing_abstract_count':1,'parser_version':2}
|
||||
self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
||||
self.plan = {'format_version':1,'inputs':[self.input(self.path)],'assertions':[
|
||||
{'source':'dblp','external_id':self.row['external_id'],'record_sha256':digest(self.raw),
|
||||
'identifiers':{'arxiv':'1408.1485'},'evidence':[{
|
||||
'file':str(self.evidence),'sha256':digest(self.evidence.read_bytes()),
|
||||
'url':'https://example.org/archive','checked_at':'2026-11-08T08:00:00Z',
|
||||
'note':'Reviewed title, authors, publication and version evidence'}]}]}
|
||||
|
||||
def input(self, path):
|
||||
return {'file':str(path),'sha256':digest(path.read_bytes()),
|
||||
'manifest_sha256':digest(path.with_suffix('.manifest.json').read_bytes())}
|
||||
|
||||
def run_plan(self, plan=None):
|
||||
path = self.root / 'plan.json'
|
||||
path.write_bytes(json_bytes(self.plan if plan is None else plan))
|
||||
return prepare(path, self.output)
|
||||
|
||||
def test_only_identifiers_and_provenance_change_and_raw_inputs_survive(self):
|
||||
result = self.run_plan()
|
||||
self.assertEqual(self.path.read_bytes(), self.raw)
|
||||
prepared = json.loads((self.output / self.path.name).read_bytes())
|
||||
provenance = prepared['metadata'].pop('identifier_curation')
|
||||
self.assertEqual(prepared.pop('identifiers'), {'arxiv':'1408.1485'})
|
||||
self.assertEqual(prepared, self.row)
|
||||
self.assertEqual(provenance[0]['record_sha256'], digest(self.raw))
|
||||
self.assertEqual((self.output / provenance[0]['evidence'][0]['file']).read_bytes(), self.evidence.read_bytes())
|
||||
manifest = json.loads((self.output / 'records.manifest.json').read_bytes())
|
||||
for key, value in self.manifest.items():
|
||||
if key != 'file_sha256': self.assertEqual(manifest[key], value)
|
||||
self.assertEqual(manifest['file_sha256'], digest((self.output / self.path.name).read_bytes()))
|
||||
self.assertEqual(result['assertions_applied'], 1)
|
||||
self.assertEqual(result['identifiers_added'], 1)
|
||||
self.assertEqual(result['verification_scope'], SCOPE)
|
||||
self.assertFalse(result['semantic_identity_verified'])
|
||||
self.assertFalse(result['database_written'])
|
||||
self.assertEqual(list(self.output.glob('*.jsonl')), [self.output / self.path.name])
|
||||
with self.assertRaisesRegex(ValueError, 'already exist'): self.run_plan()
|
||||
|
||||
def test_late_invalid_file_publishes_nothing(self):
|
||||
second = self.root / 'second.jsonl'
|
||||
second.write_bytes(json_bytes({**self.row,'external_id':'second'}))
|
||||
manifest = {**self.manifest,'file_sha256':digest(second.read_bytes()),'count':2}
|
||||
second.with_suffix('.manifest.json').write_bytes(json_bytes(manifest))
|
||||
self.plan['inputs'].append(self.input(second))
|
||||
with self.assertRaisesRegex(ValueError, 'count mismatch'): self.run_plan()
|
||||
self.assertFalse(self.output.exists())
|
||||
self.assertFalse(list(self.root.glob('.identifier-curation-*')))
|
||||
self.assertEqual(self.path.read_bytes(), self.raw)
|
||||
|
||||
def test_unasserted_rows_keep_exact_physical_bytes_and_doi_is_additive(self):
|
||||
untouched = (json.dumps({**self.row,'external_id':'unselected'},ensure_ascii=False,indent=None) + ' ').encode()
|
||||
self.path.write_bytes(self.raw + untouched)
|
||||
self.path.with_suffix('.manifest.json').write_bytes(json_bytes({**self.manifest,
|
||||
'file_sha256':digest(self.path.read_bytes()),'count':2,'missing_abstract_count':2}))
|
||||
self.plan['inputs'] = [self.input(self.path)]
|
||||
self.plan['assertions'][0]['identifiers']['doi'] = '10.1234/example'
|
||||
result = self.run_plan()
|
||||
with (self.output / self.path.name).open('rb') as stream:
|
||||
first = json.loads(next(stream))
|
||||
self.assertEqual(next(stream), untouched)
|
||||
self.assertEqual(first['identifiers'], {'arxiv':'1408.1485','doi':'10.1234/example'})
|
||||
self.assertEqual(result['identifiers_added'], 2)
|
||||
|
||||
def test_missing_duplicate_and_mismatched_assertions_fail_closed(self):
|
||||
for mode in ('missing','duplicate','line_hash','duplicate_input'):
|
||||
with self.subTest(mode=mode):
|
||||
plan = copy.deepcopy(self.plan)
|
||||
assertion = plan['assertions'][0]
|
||||
if mode == 'missing': assertion['external_id'] = 'absent'
|
||||
elif mode == 'duplicate': plan['assertions'].append(copy.deepcopy(assertion))
|
||||
elif mode == 'line_hash': assertion['record_sha256'] = '0' * 64
|
||||
else:
|
||||
other = self.root / 'duplicate.jsonl'
|
||||
other.write_bytes(self.raw)
|
||||
other.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
||||
plan['inputs'].append(self.input(other))
|
||||
with self.assertRaises(ValueError): self.run_plan(plan)
|
||||
self.assertFalse(self.output.exists())
|
||||
|
||||
def test_identifier_conflict_does_not_overwrite_and_same_value_is_auditable(self):
|
||||
self.row['identifiers'] = {'arxiv':'cs/0307069'}
|
||||
self.path.write_bytes(json_bytes(self.row))
|
||||
self.manifest['file_sha256'] = digest(self.path.read_bytes())
|
||||
self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
||||
self.plan['inputs'] = [self.input(self.path)]
|
||||
self.plan['assertions'][0]['record_sha256'] = digest(self.path.read_bytes())
|
||||
with self.assertRaisesRegex(ValueError, 'identifier conflict'): self.run_plan()
|
||||
self.assertFalse(self.output.exists())
|
||||
self.plan['assertions'][0]['identifiers']['arxiv'] = 'cs/0307069'
|
||||
self.assertEqual(self.run_plan()['identifiers_added'], 0)
|
||||
|
||||
def test_unsupported_changes_identifiers_and_unbound_evidence_are_rejected(self):
|
||||
for mode in ('title','bad_month','version','unknown_scheme','doi_case','evidence_hash','credentials','timestamp'):
|
||||
with self.subTest(mode=mode):
|
||||
plan = copy.deepcopy(self.plan)
|
||||
assertion = plan['assertions'][0]
|
||||
evidence = assertion['evidence'][0]
|
||||
if mode == 'title': assertion['title'] = 'Rewrite forbidden'
|
||||
elif mode == 'bad_month': assertion['identifiers'] = {'arxiv':'1413.12345'}
|
||||
elif mode == 'version': assertion['identifiers'] = {'arxiv':'1408.1485v2'}
|
||||
elif mode == 'unknown_scheme': assertion['identifiers'] = {'url':'https://example.org'}
|
||||
elif mode == 'doi_case': assertion['identifiers'] = {'doi':'10.1234/UPPER'}
|
||||
elif mode == 'evidence_hash': evidence['sha256'] = '0' * 64
|
||||
elif mode == 'credentials': evidence['url'] = 'https://user:secret@example.org'
|
||||
else: evidence['checked_at'] = '2026-11-08T08:00:00'
|
||||
with self.assertRaises(ValueError): self.run_plan(plan)
|
||||
self.assertFalse(self.output.exists())
|
||||
|
||||
def test_input_and_manifest_hashes_and_duplicate_json_keys_are_checked(self):
|
||||
for mode in ('input','manifest','duplicate_json','invalid_evidence_json'):
|
||||
with self.subTest(mode=mode):
|
||||
plan = copy.deepcopy(self.plan)
|
||||
if mode == 'input':
|
||||
plan['inputs'][0]['sha256'] = '0' * 64
|
||||
elif mode == 'manifest':
|
||||
plan['inputs'][0]['manifest_sha256'] = '0' * 64
|
||||
elif mode == 'duplicate_json':
|
||||
raw = self.raw[:-2] + b',"source":"dblp"}\n'
|
||||
self.path.write_bytes(raw)
|
||||
self.path.with_suffix('.manifest.json').write_bytes(json_bytes({**self.manifest,'file_sha256':digest(raw)}))
|
||||
plan['inputs'] = [self.input(self.path)]
|
||||
else:
|
||||
self.evidence.write_bytes(b'not json')
|
||||
plan['assertions'][0]['evidence'][0]['sha256'] = digest(self.evidence.read_bytes())
|
||||
with self.assertRaises(ValueError): self.run_plan(plan)
|
||||
self.assertFalse(self.output.exists())
|
||||
self.path.write_bytes(self.raw)
|
||||
self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
||||
|
||||
def test_curated_anchor_import_does_not_merge_extended_version_or_its_history(self):
|
||||
db = self.root / 'papers.sqlite'
|
||||
store = Store(db)
|
||||
try:
|
||||
original, _ = store.record('arxiv','1408.1485','https://example.org/one',self.row['title'],None,self.row['authors'],{'arxiv':'1408.1485'})
|
||||
extended, record = store.record('arxiv','cs/0307069','https://example.org/two',self.row['title'],None,self.row['authors'],{'arxiv':'cs/0307069'})
|
||||
store.decision(extended,record,'ICLR',2025,'main','another-attempt','https://example.org/reject','rejected','Reject')
|
||||
store.conn.commit()
|
||||
finally:
|
||||
store.conn.close()
|
||||
self.run_plan()
|
||||
ingest_file(db,self.output / self.path.name)
|
||||
store = Store(db)
|
||||
try:
|
||||
c = store.conn
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
|
||||
self.assertEqual(c.execute('SELECT paper_id FROM source_records WHERE external_id=?',(self.row['external_id'],)).fetchone()[0], original)
|
||||
self.assertEqual(c.execute('SELECT s.paper_id FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE v.acronym=?',('ICLR',)).fetchone()[0], extended)
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0], 0)
|
||||
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
|
||||
finally:
|
||||
store.conn.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,116 @@
|
||||
from contextlib import redirect_stdout
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from paper_atlas.ingestion.records import ingest_file
|
||||
|
||||
|
||||
class IncrementalCoverageTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.db = self.root / 'papers.sqlite'
|
||||
|
||||
def file(self, name, external_ids, **options):
|
||||
rows = [{'source':'example','external_id':key,'url':'https://example.org/'+key,
|
||||
'venue':'Example','year':2026,'track':'main','status':'accepted',
|
||||
'title':'Independent research contribution '+key,'abstract':'Known abstract',
|
||||
'authors':[], 'identifiers':{'doi':'10.1234/'+key}} for key in external_ids]
|
||||
path = self.root / (name+'.jsonl')
|
||||
path.write_bytes((''.join(json.dumps(row)+'\n' for row in rows)).encode())
|
||||
manifest = {key:rows[0][key] for key in ('source','venue','year','track')}
|
||||
manifest.update(list_url='https://example.org/2026', count=len(rows),status='partial',
|
||||
file_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
|
||||
manifest.update(options)
|
||||
path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
|
||||
return path
|
||||
|
||||
def ingest(self, path):
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ingest_file(self.db,path)
|
||||
|
||||
def incremental(self, previous=2, prior_expected=2, expected=4):
|
||||
return {'coverage_mode':'incremental','coverage_previous_count':previous,
|
||||
'coverage_previous_expected_count':prior_expected,'coverage_expected_count':expected}
|
||||
|
||||
def coverage(self):
|
||||
with sqlite3.connect(self.db) as c:
|
||||
return c.execute('SELECT status,expected_count,imported_count FROM edition_sources').fetchall()
|
||||
|
||||
def test_append_and_refresh_count_actual_source_records_without_erasing_earlier_batch(self):
|
||||
self.ingest(self.file('first',['a','b']))
|
||||
self.ingest(self.file('append',['c'],**self.incremental()))
|
||||
self.assertEqual(self.coverage(),[('partial',4,3)])
|
||||
self.ingest(self.file('refresh',['c'],**self.incremental(3,4,4)))
|
||||
self.assertEqual(self.coverage(),[('partial',4,3)])
|
||||
with sqlite3.connect(self.db) as c:
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],3)
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],3)
|
||||
runs = c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id').fetchall()
|
||||
self.assertEqual([r[0] for r in runs],[2,1,1])
|
||||
self.assertEqual(json.loads(runs[-1][1])['coverage'],
|
||||
{'mode':'incremental','previous_count':3,'expected_count':4,'imported_count':3})
|
||||
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
|
||||
|
||||
def test_stale_parent_stops_before_source_writes(self):
|
||||
self.ingest(self.file('first',['a','b']))
|
||||
with self.assertRaisesRegex(ValueError,'parent changed'):
|
||||
self.ingest(self.file('stale',['c'],**self.incremental(1,2,4)))
|
||||
self.assertEqual(self.coverage(),[('partial',2,2)])
|
||||
with sqlite3.connect(self.db) as c:
|
||||
self.assertEqual(c.execute("SELECT count(*) FROM source_records WHERE external_id='c'").fetchone()[0],0)
|
||||
self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
||||
|
||||
def test_late_source_bound_failure_rolls_back_new_records_and_coverage(self):
|
||||
self.ingest(self.file('first',['a','b']))
|
||||
with self.assertRaisesRegex(ValueError,'source bounds'):
|
||||
self.ingest(self.file('overflow',['c'],**self.incremental(2,2,2)))
|
||||
self.assertEqual(self.coverage(),[('partial',2,2)])
|
||||
with sqlite3.connect(self.db) as c:
|
||||
for table in ('papers','source_records','submissions','decisions'):
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],2)
|
||||
|
||||
def test_multiple_listings_cannot_be_silently_aggregated(self):
|
||||
self.ingest(self.file('one',['a']))
|
||||
self.ingest(self.file('two',['b'],list_url='https://example.org/another-list'))
|
||||
before=self.coverage()
|
||||
with self.assertRaisesRegex(ValueError,'unambiguous source listing'):
|
||||
self.ingest(self.file('ambiguous',['c'],**self.incremental(2,1,4)))
|
||||
self.assertEqual(self.coverage(),before)
|
||||
|
||||
def test_untracked_source_rows_require_reconciliation(self):
|
||||
self.ingest(self.file('first',['a','b']))
|
||||
with sqlite3.connect(self.db) as c:
|
||||
c.execute('UPDATE edition_sources SET imported_count=1')
|
||||
with self.assertRaisesRegex(ValueError,'actual source records'):
|
||||
self.ingest(self.file('stale-summary',['c'],**self.incremental(1,2,4)))
|
||||
self.assertEqual(self.coverage(),[('partial',2,1)])
|
||||
|
||||
def test_first_partial_chunk_can_declare_an_independently_known_source_total(self):
|
||||
self.ingest(self.file('first',['a'],**self.incremental(0,None,4)))
|
||||
self.assertEqual(self.coverage(),[('partial',4,1)])
|
||||
|
||||
def test_complete_listing_is_not_demoted_by_a_partial_chunk(self):
|
||||
self.ingest(self.file('complete',['a','b'],status='complete'))
|
||||
with self.assertRaisesRegex(ValueError,'not partial'):
|
||||
self.ingest(self.file('append',['c'],**self.incremental()))
|
||||
self.assertEqual(self.coverage(),[('complete',2,2)])
|
||||
|
||||
def test_invalid_mode_counts_and_missing_scope_fail_before_database_creation(self):
|
||||
for options in ({'coverage_mode':'unknown'}, self.incremental(True,2,4),
|
||||
self.incremental(2,5,4),self.incremental(-1,2,4),
|
||||
{'coverage_mode':'incremental'},
|
||||
{**self.incremental(),'status':'complete'}):
|
||||
with self.subTest(options=options):
|
||||
with self.assertRaises(ValueError):self.ingest(self.file('invalid',['a'],**options))
|
||||
self.assertFalse(self.db.exists())
|
||||
|
||||
|
||||
if __name__=='__main__':
|
||||
unittest.main()
|
||||
@@ -53,6 +53,50 @@ class StoreTests(unittest.TestCase):
|
||||
self.assertNotEqual(other,paper)
|
||||
self.assertEqual(self.store.conn.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1)
|
||||
|
||||
def test_identifier_anchor_does_not_absorb_a_same_title_arxiv_version(self):
|
||||
conference,first=self.record('arxiv','1408.1485')
|
||||
extended,second=self.record('arxiv','cs/0307069')
|
||||
self.store.decision(extended,second,'ICLR',2025,'main','separate','https://example.org/reject','rejected','Reject')
|
||||
resolved,_=self.record('dblp','conf/uai/example',{'arxiv':'1408.1485'})
|
||||
self.assertEqual(resolved,conference);self.assertNotEqual(conference,extended)
|
||||
c=self.store.conn
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2)
|
||||
self.assertEqual(c.execute('SELECT paper_id FROM submissions').fetchone()[0],extended)
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],0)
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1)
|
||||
# A refresh of the anchored original source is equally unambiguous.
|
||||
self.assertEqual(self.record('arxiv','1408.1485')[0],conference)
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2)
|
||||
|
||||
def test_ambiguous_title_aborts_whole_file_without_merging_existing_entities(self):
|
||||
self.record('arxiv','1408.1485');self.record('arxiv','cs/0307069')
|
||||
self.store.conn.commit()
|
||||
rows=[{'source':'dblp','external_id':'conf/uai/new','url':'https://example.org/new',
|
||||
'title':'An unrelated earlier record in this file','abstract':None,'authors':[],
|
||||
'venue':'UAI','year':2001,'track':'main','status':'accepted'},
|
||||
{'source':'dblp','external_id':'conf/uai/ambiguous','url':'https://example.org/ambiguous',
|
||||
'title':'A sufficiently long example research paper title','abstract':None,
|
||||
'authors':['Jane Doe','John Smith'],'identifiers':{'doi':'10.0000/not-yet-known'},
|
||||
'venue':'UAI','year':2001,'track':'main','status':'accepted'}]
|
||||
path=Path(self.tmp.name)/'ambiguous.jsonl'
|
||||
path.write_text(''.join(json.dumps(row)+'\n' for row in rows))
|
||||
path.with_suffix('.manifest.json').write_text(json.dumps({'source':'dblp','list_url':'https://example.org/list','count':2}))
|
||||
with self.assertRaisesRegex(ValueError,'Ambiguous paper identity'):ingest_file(self.db,path)
|
||||
c=self.store.conn
|
||||
for table,count in [('papers',2),('source_records',2),('submissions',0),('decisions',0),('paper_merges',0)]:
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],count,table)
|
||||
self.assertEqual(c.execute('SELECT status FROM ingest_runs').fetchone()[0],'failed')
|
||||
|
||||
def test_explicit_bridge_merges_only_its_identifier_matches(self):
|
||||
first,_=self.record('arxiv','2601.00001',authors=['First Author'])
|
||||
second,_=self.record('arxiv','2601.00002',{'doi':'10.0000/bridge'},authors=['Second Author'])
|
||||
unrelated,_=self.record('arxiv','2601.00003')
|
||||
resolved,_=self.record('official','bridge',{'arxiv':'2601.00001','doi':'10.0000/bridge'})
|
||||
self.assertEqual(resolved,min(first,second))
|
||||
self.assertEqual(self.store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],2)
|
||||
self.assertIsNotNone(self.store.conn.execute('SELECT 1 FROM papers WHERE id=?',(unrelated,)).fetchone())
|
||||
self.assertEqual(self.store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1)
|
||||
|
||||
def test_manifest_count_mismatch_rolls_back_records(self):
|
||||
path=Path(self.tmp.name)/'source.jsonl'
|
||||
path.write_text(json.dumps({'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'})+'\n')
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
from contextlib import closing, redirect_stdout
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import uuid
|
||||
|
||||
from paper_atlas.collectors.dblp import XML_SHA
|
||||
from paper_atlas.collectors.dblp_scope import scope
|
||||
from paper_atlas.collectors.dblp_records import normalize_row
|
||||
from paper_atlas.db.store import Store
|
||||
from paper_atlas.db.backend import open_store
|
||||
from paper_atlas.ingestion.joint_editions import (TARGET, MARKER, inspect, apply_plan, run)
|
||||
from paper_atlas.postgres.client import connect
|
||||
from paper_atlas.postgres.migration import migrate
|
||||
from paper_atlas.postgres.schema import quote
|
||||
from paper_atlas import querying
|
||||
|
||||
|
||||
def seed(store):
|
||||
c = store.conn
|
||||
c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,'reasoning','conference')", (TARGET,TARGET))
|
||||
target = c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0]
|
||||
for alias in ['ECAI','IJCAI']:
|
||||
c.execute('INSERT INTO venue_aliases VALUES(?,?)',(target,alias))
|
||||
for year in [2018,2019,2022]:
|
||||
metadata = {'snapshot_sha256':XML_SHA,'proceedings_key':'conf/ijcai/'+str(year),
|
||||
'bibliographic_year':year,'bibliography':{'booktitle':['IJCAI']}}
|
||||
p,r=store.record('dblp','conf/ijcai/fixture'+str(year),'https://dblp.org/rec/fixture',
|
||||
'Original publication '+str(year),'Abstract\x00with\x01data',[],
|
||||
{'doi':'10.24963/ijcai.'+str(year)+'/1'}, {})
|
||||
store.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'IJCAI'}])
|
||||
store.decision(p,r,'IJCAI',year,'main','fixture'+str(year),'https://example.org/decision','accepted','listed')
|
||||
stored=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r,)).fetchone()[0]);stored.update(metadata)
|
||||
c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(stored),r))
|
||||
c.commit()
|
||||
|
||||
|
||||
def plan(c):
|
||||
return {**inspect(c),'database':{'engine':'sqlite','path':'fixture'}}
|
||||
|
||||
|
||||
class JointEditionTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
|
||||
self.db=Path(self.tmp.name)/'sample.sqlite'
|
||||
self.store=Store(self.db); self.addCleanup(self.store.conn.close)
|
||||
seed(self.store); self.c=self.store.conn
|
||||
|
||||
def test_apply_preserves_every_protected_row_and_edition_id(self):
|
||||
before=plan(self.c); ids=[r['id']for r in before['editions']]
|
||||
result=apply_plan(self.c,before);self.c.commit()
|
||||
self.assertEqual(result['updated_editions'],2)
|
||||
self.assertEqual(result['content_fingerprints'],before['content_fingerprints'])
|
||||
self.assertEqual([r['id']for r in result['editions']],ids)
|
||||
untouched=self.c.execute('SELECT v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id WHERE e.year=2019').fetchone()[0]
|
||||
self.assertEqual(untouched,'IJCAI')
|
||||
self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(),[])
|
||||
self.assertEqual(self.c.execute('SELECT count(*) FROM schema_migrations WHERE name=?',(MARKER,)).fetchone()[0],1)
|
||||
with redirect_stdout(io.StringIO())as out:
|
||||
querying.main(['--db',str(self.db),'--venue','ECAI'])
|
||||
self.assertEqual(len(json.loads(out.getvalue())),2)
|
||||
self.assertEqual(apply_plan(self.c,before)['updated_editions'],0)
|
||||
|
||||
def test_refresh_of_old_normalized_source_does_not_recreate_ordinary_edition(self):
|
||||
before=plan(self.c);apply_plan(self.c,before)
|
||||
for year in [2018,2022]:
|
||||
row=self.c.execute('''SELECT r.paper_id,r.id,d.observed_at FROM source_records r
|
||||
JOIN submissions s ON s.source_record_id=r.id JOIN decisions d ON d.submission_id=s.id
|
||||
WHERE r.external_id=?''',('conf/ijcai/fixture'+str(year),)).fetchone()
|
||||
self.store.decision(row[0],row[1],'IJCAI',year,'main','fixture'+str(year),
|
||||
'https://example.org/decision','accepted','listed',observed_at=row[2])
|
||||
self.assertEqual(inspect(self.c)['content_fingerprints'],before['content_fingerprints'])
|
||||
self.assertEqual(self.c.execute('SELECT count(*) FROM editions').fetchone()[0],3)
|
||||
|
||||
def test_any_changed_source_text_rejects_stale_plan_before_membership_updates(self):
|
||||
before=plan(self.c)
|
||||
self.c.execute("UPDATE source_records SET title='New evidence' WHERE external_id='conf/ijcai/fixture2018'")
|
||||
with self.assertRaisesRegex(ValueError,'Stale repair plan'):
|
||||
apply_plan(self.c,before)
|
||||
self.assertEqual(inspect(self.c)['status'],'ready')
|
||||
|
||||
def test_conflicting_target_edition_is_never_merged_or_deleted(self):
|
||||
target=self.c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0]
|
||||
self.c.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2018,'main')",(target,))
|
||||
with self.assertRaisesRegex(ValueError,'colliding'):
|
||||
inspect(self.c)
|
||||
self.assertEqual(self.c.execute('SELECT count(*) FROM editions WHERE year=2018').fetchone()[0],2)
|
||||
|
||||
def test_wrong_source_volume_and_snapshot_fail_closed(self):
|
||||
original=self.c.execute("SELECT metadata_json FROM source_records WHERE external_id='conf/ijcai/fixture2018'").fetchone()[0]
|
||||
for key,value in [('proceedings_key','conf/ijcai/2018w'),('snapshot_sha256','wrong'),('bibliographic_year',2017),('bibliography',{'booktitle':['Other']})]:
|
||||
m=json.loads(original);m[key]=value
|
||||
self.c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='conf/ijcai/fixture2018'",(json.dumps(m),))
|
||||
with self.subTest(key=key),self.assertRaisesRegex(ValueError,'Source evidence'):
|
||||
inspect(self.c)
|
||||
|
||||
def test_missing_source_support_and_wrong_aliases_are_not_guessed(self):
|
||||
self.c.execute("DELETE FROM venue_aliases WHERE alias='ECAI'")
|
||||
with self.assertRaisesRegex(ValueError,'aliases'):
|
||||
inspect(self.c)
|
||||
self.c.rollback()
|
||||
self.c.execute('DELETE FROM submission_sources WHERE source_record_id=(SELECT id FROM source_records WHERE external_id=?)',('conf/ijcai/fixture2022',))
|
||||
with self.assertRaisesRegex(ValueError,'Every submission'):
|
||||
inspect(self.c)
|
||||
|
||||
def test_cli_plan_readonly_hash_binding_identity_and_repeat_apply(self):
|
||||
out=Path(self.tmp.name)/'plan.json'
|
||||
with redirect_stdout(io.StringIO()):run(self.db,out)
|
||||
self.assertEqual(inspect(self.c)['status'],'ready')
|
||||
raw=out.read_bytes();sha=hashlib.sha256(raw).hexdigest()
|
||||
with patch('paper_atlas.ingestion.joint_editions.open_store')as writer:
|
||||
with self.assertRaisesRegex(ValueError,'hash differs'):
|
||||
run(self.db,Path(self.tmp.name)/'bad.json',plan=out,plan_sha256='0'*64)
|
||||
writer.assert_not_called()
|
||||
with redirect_stdout(io.StringIO()):
|
||||
result=run(self.db,Path(self.tmp.name)/'applied.json',plan=out,plan_sha256=sha)
|
||||
again=run(self.db,Path(self.tmp.name)/'retry.json',plan=out,plan_sha256=sha)
|
||||
self.assertEqual(result['updated_editions'],2);self.assertEqual(again['updated_editions'],0)
|
||||
self.assertEqual(result['plan_sha256'],sha);self.assertEqual(again['plan_sha256'],sha)
|
||||
altered=json.loads(raw);altered['database']['path']='another.sqlite'
|
||||
bad=Path(self.tmp.name)/'wrong-db.json';bad.write_text(json.dumps(altered))
|
||||
with self.assertRaisesRegex(ValueError,'another database'):
|
||||
run(self.db,Path(self.tmp.name)/'wrong-result.json',plan=bad,plan_sha256=hashlib.sha256(bad.read_bytes()).hexdigest())
|
||||
|
||||
def test_invalid_empty_plan_fails_before_opening_writer(self):
|
||||
f=Path(self.tmp.name)/'empty.json';f.write_text('{}')
|
||||
with patch('paper_atlas.ingestion.joint_editions.open_store')as writer:
|
||||
with self.assertRaises(ValueError):
|
||||
run(self.db,Path(self.tmp.name)/'empty-result.json',plan=f,plan_sha256=hashlib.sha256(f.read_bytes()).hexdigest())
|
||||
writer.assert_not_called()
|
||||
|
||||
def test_collector_only_changes_the_two_exact_main_volume_keys(self):
|
||||
for year in [2018,2019,2022]:
|
||||
volume={'key':'conf/ijcai/'+str(year),'fields':{'title':['IJCAI '+str(year)],'year':[str(year)]}}
|
||||
row={'record_type':'inproceedings','key':'conf/ijcai/One'+str(year),'fields':{'booktitle':['IJCAI'],
|
||||
'year':[str(year)],'crossref':[volume['key']],'title':['Original research title'],
|
||||
'author':['Jane Doe'],'ee':['https://doi.org/10.24963/ijcai.'+str(year)+'/1']}}
|
||||
result,error=normalize_row(row,{volume['key']:volume},{'IJCAI','ECAI'})
|
||||
self.assertIsNone(error)
|
||||
self.assertEqual(result['venue'],TARGET if year in(2018,2022)else'IJCAI')
|
||||
self.assertEqual(result['metadata']['bibliography']['booktitle'],['IJCAI'])
|
||||
self.assertEqual(result['title'],'Original research title')
|
||||
if year in(2018,2022):
|
||||
self.assertIn('joint_edition_evidence',result['metadata'])
|
||||
row['fields']['booktitle']=['IJCAI (Workshop)']
|
||||
scoped,_=scope(row,volume)
|
||||
self.assertEqual(scoped['venue'],'IJCAI')
|
||||
self.assertEqual(scoped['track'],'workshop')
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service')
|
||||
class PostgreSQLJointTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.target='pa_joint_'+uuid.uuid4().hex[:12]
|
||||
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
|
||||
self.db=Path(self.tmp.name)/'sample.sqlite';s=Store(self.db);seed(s);s.conn.close()
|
||||
self.addCleanup(self.cleanup)
|
||||
with redirect_stdout(io.StringIO()):migrate(self.db,self.service,Path(self.tmp.name)/'migration.json',schema=self.target,min_free_gib=0)
|
||||
|
||||
def cleanup(self):
|
||||
with connect(self.service)as c:
|
||||
for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)',(self.target,)).fetchall():
|
||||
c.execute('DROP SCHEMA '+quote(name)+' CASCADE')
|
||||
|
||||
def test_native_repair_is_idempotent_keeps_codec_and_provenance(self):
|
||||
p=Path(self.tmp.name)/'plan.json'
|
||||
with redirect_stdout(io.StringIO()):
|
||||
before=run(None,p,pg_service=self.service,pg_schema=self.target)
|
||||
after=run(None,Path(self.tmp.name)/'applied.json',pg_service=self.service,pg_schema=self.target,plan=p,plan_sha256=hashlib.sha256(p.read_bytes()).hexdigest())
|
||||
self.assertEqual(after['content_fingerprints'],before['content_fingerprints'])
|
||||
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c:
|
||||
self.assertEqual(apply_plan(c,before)['updated_editions'],0)
|
||||
self.assertEqual(c.execute('SELECT abstract FROM papers ORDER BY id LIMIT 1').fetchone()[0],'Abstract\x00with\x01data')
|
||||
|
||||
def test_native_stale_source_plan_preserves_both_original_memberships(self):
|
||||
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c:
|
||||
before=plan(c)
|
||||
c.execute("UPDATE source_records SET title='Changed' WHERE external_id=?",('conf/ijcai/fixture2022',));c.commit()
|
||||
with self.assertRaisesRegex(ValueError,'Stale repair plan'):
|
||||
apply_plan(c,before)
|
||||
c.rollback()
|
||||
self.assertEqual(inspect(c)['status'],'ready')
|
||||
@@ -175,6 +175,22 @@ class PostgreSQLIntegrationTests(unittest.TestCase):
|
||||
finally:
|
||||
reader.close()
|
||||
|
||||
def test_history_audit_matches_sqlite_for_actual_postgresql_snapshot(self):
|
||||
from paper_atlas.audits.histories import report
|
||||
from tests.test_histories import add_history_fixture
|
||||
store = Store(self.db)
|
||||
try:
|
||||
add_history_fixture(store)
|
||||
finally:
|
||||
store.conn.close()
|
||||
expected = report(self.db)
|
||||
self.assertEqual(self.migrate()['status'], 'complete')
|
||||
actual = report(pg_service=self.service, pg_schema=self.target)
|
||||
self.assertEqual(actual['engine'], 'postgresql')
|
||||
self.assertEqual(actual['schema'], self.target)
|
||||
for key in ('counts', 'latest_attempt_status_counts', 'evidence_relationship_checks', 'examples'):
|
||||
self.assertEqual(actual[key], expected[key], key)
|
||||
|
||||
def interrupt_after_first_table(self):
|
||||
calls = 0
|
||||
def interrupt(*args):
|
||||
|
||||
@@ -87,6 +87,82 @@ class PostgreSQLIngestionTests(unittest.TestCase):
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ingest_file(None, path, pg_service=self.service, pg_schema=self.schema)
|
||||
|
||||
def test_incremental_source_coverage_counts_prior_chunks_and_refreshes_on_postgres(self):
|
||||
rows = [self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
|
||||
for key in ('incremental1','incremental2','incremental3')]
|
||||
listing = 'https://example.org/uai/2001'
|
||||
def manifest(path, **options):
|
||||
target=path.with_suffix('.manifest.json')
|
||||
value=json.loads(target.read_text())
|
||||
value.update(list_url=listing,status='partial',**options)
|
||||
target.write_text(json.dumps(value))
|
||||
first=self.file('initial',rows[:2]);manifest(first);self.ingest(first)
|
||||
second=self.file('append',[rows[2]])
|
||||
manifest(second,coverage_mode='incremental',coverage_previous_count=2,
|
||||
coverage_previous_expected_count=2,coverage_expected_count=4)
|
||||
self.ingest(second)
|
||||
manifest(second,coverage_mode='incremental',coverage_previous_count=3,
|
||||
coverage_previous_expected_count=4,coverage_expected_count=4)
|
||||
self.ingest(second)
|
||||
with closing(ReadConnection(self.service,self.schema)) as c:
|
||||
coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
|
||||
self.assertEqual(tuple(coverage.values),('partial',4,3))
|
||||
stored=c.execute('SELECT abstract FROM source_records WHERE external_id=?',('incremental3',)).fetchone()
|
||||
self.assertEqual(stored[0],rows[2]['abstract'])
|
||||
self.assertEqual(c.execute("SELECT count(*) FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name='dblp'").fetchone()[0],3)
|
||||
run=c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()
|
||||
self.assertEqual(run[0],1)
|
||||
self.assertEqual(json.loads(run[1])['coverage']['imported_count'],3)
|
||||
|
||||
def test_incremental_postgres_bound_failure_rolls_back_rows_and_summary(self):
|
||||
listing='https://example.org/uai/2001'
|
||||
rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
|
||||
for key in ('bound1','bound2','bound3')]
|
||||
first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
|
||||
value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
|
||||
self.ingest(first)
|
||||
with closing(ReadConnection(self.service,self.schema)) as c:
|
||||
before={table:c.execute('SELECT count(*) FROM '+table).fetchone()[0]
|
||||
for table in ('papers','source_records','submissions','decisions')}
|
||||
path=self.file('overflow',[rows[2]]);target=path.with_suffix('.manifest.json')
|
||||
value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
|
||||
coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=2)
|
||||
target.write_text(json.dumps(value))
|
||||
with self.assertRaisesRegex(ValueError,'source bounds'):self.ingest(path)
|
||||
with closing(ReadConnection(self.service,self.schema)) as c:
|
||||
self.assertEqual({table:c.execute('SELECT count(*) FROM '+table).fetchone()[0] for table in before},before)
|
||||
coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
|
||||
self.assertEqual(tuple(coverage.values),('partial',2,2))
|
||||
self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
||||
|
||||
def test_incremental_postgres_does_not_overwrite_a_concurrently_changed_parent(self):
|
||||
listing='https://example.org/uai/2001'
|
||||
rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
|
||||
for key in ('concurrent1','concurrent2','concurrent3')]
|
||||
first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
|
||||
value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
|
||||
self.ingest(first)
|
||||
path=self.file('racing',[rows[2]]);target=path.with_suffix('.manifest.json')
|
||||
value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
|
||||
coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=4)
|
||||
target.write_text(json.dumps(value))
|
||||
original=Store.record
|
||||
def change_parent(store,*args,**kwargs):
|
||||
result=original(store,*args,**kwargs)
|
||||
# A separate committed SQL writer need not honor the native
|
||||
# advisory lock. Its update must survive our rejected file.
|
||||
with connect(self.service) as other:
|
||||
other.execute('SET search_path TO '+quote(self.schema)+',pg_catalog')
|
||||
other.execute('UPDATE edition_sources SET expected_count=99 WHERE list_url=%s',(listing,))
|
||||
return result
|
||||
with patch.object(Store,'record',change_parent):
|
||||
with self.assertRaisesRegex(ValueError,'parent changed during import'):self.ingest(path)
|
||||
with closing(ReadConnection(self.service,self.schema)) as c:
|
||||
coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
|
||||
self.assertEqual(tuple(coverage.values),('partial',99,2))
|
||||
self.assertEqual(c.execute('SELECT count(*) FROM source_records WHERE external_id=?',('concurrent3',)).fetchone()[0],0)
|
||||
self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
||||
|
||||
def test_native_writes_keep_nul_historical_ids_and_later_acceptance(self):
|
||||
store = self.store()
|
||||
try:
|
||||
@@ -110,6 +186,34 @@ class PostgreSQLIngestionTests(unittest.TestCase):
|
||||
finally:
|
||||
reader.close()
|
||||
|
||||
def test_identity_anchor_keeps_other_arxiv_version_and_ambiguous_file_rolls_back(self):
|
||||
title='An identical long research title shared by conference and extended versions'
|
||||
authors=['Jane Doe','John Smith']
|
||||
store=self.store()
|
||||
try:
|
||||
first,_=store.record('arxiv','1408.1485','https://arxiv.org/abs/1408.1485',title,'Conference abstract',authors)
|
||||
second,_=store.record('arxiv','cs/0307069','https://arxiv.org/abs/cs/0307069',title,'Extended abstract',authors)
|
||||
resolved,_=store.record('dblp','known','https://example.org/known',title,None,authors,{'arxiv':'1408.1485'})
|
||||
self.assertEqual(resolved,first);self.assertNotEqual(first,second)
|
||||
store.conn.commit()
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],3)
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) # migration fixture only
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1)
|
||||
finally:store.conn.close()
|
||||
rows=[self.row('official','earlier',status='accepted',venue='UAI',year=2001),
|
||||
dict(self.row('official','ambiguous',status='accepted',venue='UAI',year=2001),
|
||||
title=title,abstract=None,authors=authors,identifiers={})]
|
||||
with self.assertRaisesRegex(ValueError,'Ambiguous paper identity'):self.ingest(self.file('ambiguous-identity',rows))
|
||||
store=self.store()
|
||||
try:
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],3)
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],4)
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
|
||||
self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1)
|
||||
self.assertEqual(store.conn.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
||||
self.assertEqual(store.conn.execute("SELECT count(*) FROM source_records WHERE external_id IN (?,?)",('earlier','ambiguous')).fetchone()[0],0)
|
||||
finally:store.conn.close()
|
||||
|
||||
def test_file_import_is_idempotent_and_refreshes_only_own_tags(self):
|
||||
row = self.row(); row['tags'] = [{'namespace': 'openreview_snapshot:area', 'value': 'theory'}]
|
||||
path = self.file('reject', [row]); self.ingest(path)
|
||||
@@ -243,6 +347,50 @@ class PostgreSQLIngestionTests(unittest.TestCase):
|
||||
finally:
|
||||
reader.close()
|
||||
|
||||
def test_coverage_matches_sqlite_with_history_aliases_and_missing_source_abstracts(self):
|
||||
from paper_atlas.audits.coverage import report
|
||||
from tests.test_coverage_report import add_coverage_evidence
|
||||
registry = self.root / 'venues.json'
|
||||
registry.write_text(json.dumps({'scope_note': 'Candidate scope', 'venues': [
|
||||
{'acronym': 'coling', 'field': 'language'}, {'acronym': 'AAAI', 'field': 'general_ai'}]}))
|
||||
sample = Store(self.root / 'seed.sqlite')
|
||||
try:
|
||||
add_coverage_evidence(sample)
|
||||
finally:
|
||||
sample.conn.close()
|
||||
store = self.store()
|
||||
try:
|
||||
add_coverage_evidence(store)
|
||||
finally:
|
||||
store.conn.close()
|
||||
sqlite_report = report(self.root / 'seed.sqlite', registry)
|
||||
observed = []
|
||||
def reader(*args, **kwargs):
|
||||
connection = open_read_connection(*args, **kwargs)
|
||||
observed.append((connection.execute('SHOW transaction_read_only').fetchone()[0],
|
||||
connection.execute('SHOW transaction_isolation').fetchone()[0]))
|
||||
return connection
|
||||
with patch('paper_atlas.audits.coverage.open_read_connection', side_effect=reader):
|
||||
actual = report(None, registry, pg_service=self.service, pg_schema=self.schema)
|
||||
self.assertEqual(observed, [('on', 'repeatable read')])
|
||||
self.assertEqual(actual['engine'], 'postgresql')
|
||||
self.assertEqual(actual['schema'], self.schema)
|
||||
self.assertEqual(actual['pg_service'], self.service)
|
||||
def content(result):
|
||||
values = json.loads(json.dumps(result['venues']))
|
||||
for venue in values:
|
||||
for edition in venue['editions']:
|
||||
for key in ('id', 'venue_id'):
|
||||
edition.pop(key)
|
||||
return values
|
||||
self.assertEqual(content(actual), content(sqlite_report))
|
||||
self.assertEqual([r['acronym'] for r in actual['imported_venues_outside_registry']], ['ICLR'])
|
||||
edition = actual['venues'][0]['editions'][0]
|
||||
self.assertEqual(edition['latest_observed_status_counts']['accepted'], 1)
|
||||
self.assertEqual(edition['latest_observed_status_counts']['unknown'], 1)
|
||||
self.assertEqual(edition['decision_observation_counts']['rejected'], 1)
|
||||
self.assertEqual(edition['papers_without_abstract_text'], 1)
|
||||
|
||||
def test_backfill_merges_evidence_without_reimporting_or_losing_history(self):
|
||||
files = []
|
||||
for source, external, title in [('official', 'legacy', 'First bibliographic name'), ('openreview_snapshot', 'forum', 'Renamed submission')]:
|
||||
|
||||
@@ -45,10 +45,12 @@ class PublicOpenReviewTests(unittest.TestCase):
|
||||
|
||||
def test_only_explicit_decisions_and_verified_scope_become_events(self):
|
||||
for value in ('Pending', 'Active', 'Submitted to NeurIPS 2025', 'TBD', '',
|
||||
'Low score', 'ICML 2025 Retracted by Authors', 'Do not reject'):
|
||||
'Low score', 'ICML 2025 Retracted by Authors', 'Do not reject',
|
||||
'Workshop', 'Poster Workshop', 'Oral Workshop'):
|
||||
self.assertIsNone(negative_status(value), value)
|
||||
self.assertEqual(negative_status('Reject'), 'rejected')
|
||||
self.assertEqual(negative_status('Desk rejected'), 'desk_rejected')
|
||||
self.assertEqual(negative_status('Withdraw'), 'withdrawn')
|
||||
self.assertEqual(negative_status('NeurIPS 2023 Conference Withdrawn Submission'), 'withdrawn')
|
||||
for raw, expected in [('main', 'main'), ('Position', 'position'),
|
||||
('Datasets & Benchmarks', 'datasets_benchmarks')]:
|
||||
@@ -98,6 +100,121 @@ class PublicOpenReviewTests(unittest.TestCase):
|
||||
self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
|
||||
self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
|
||||
|
||||
def test_missing_title_is_held_without_losing_valid_negative_records(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
result = self.collect(tmp, {'venues/iclr/iclr2017.jsonl': [
|
||||
self.row(id='SJUdkecgx', title=''),
|
||||
self.row(id='NoTitle001', title=None),
|
||||
self.row(id='Whitespace', title=' '),
|
||||
self.row(id='ValidReject'),
|
||||
self.row(id='Withdraw01', status='Withdraw'),
|
||||
self.row(id='Workshop01', status='Workshop'),
|
||||
]}, families=['iclr'], years=[2017])
|
||||
self.assertEqual(result['input_rows'], 6)
|
||||
self.assertEqual(result['normalized_records'], 2)
|
||||
self.assertEqual(result['source_files'][0]['excluded_or_held'],
|
||||
{'missing_title': 3, 'not_explicit_negative': 1})
|
||||
held_path = Path(result['held_file'])
|
||||
self.assertEqual(hashlib.sha256(held_path.read_bytes()).hexdigest(),
|
||||
result['held_file_sha256'])
|
||||
held = [json.loads(line) for line in held_path.read_text().splitlines()]
|
||||
self.assertEqual([row['id'] for row in held], ['SJUdkecgx', 'NoTitle001', 'Whitespace'])
|
||||
self.assertTrue(all(row['reason'] == 'missing_title' and row['status'] == 'Reject'
|
||||
for row in held))
|
||||
self.assertEqual([row['provenance']['source_line'] for row in held], [1, 2, 3])
|
||||
self.assertNotIn('private@example.org', held_path.read_text())
|
||||
self.assertNotIn('private review', held_path.read_text())
|
||||
manifest = result['files'][0]
|
||||
self.assertEqual((manifest['parser_version'], manifest['status']), (3, 'partial'))
|
||||
rows = [json.loads(line) for line in Path(manifest['file']).read_text().splitlines()]
|
||||
self.assertEqual([(row['identifiers']['openreview'], row['status']) for row in rows],
|
||||
[('ValidReject', 'rejected'), ('Withdraw01', 'withdrawn')])
|
||||
for row in held:
|
||||
self.assertEqual(row['provenance']['source_revision'], REVISION)
|
||||
|
||||
def test_nontext_title_still_fails_collection(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with self.assertRaisesRegex(ValueError, 'lacks a title'):
|
||||
self.collect(tmp, {'venues/iclr/iclr2017.jsonl': [self.row(title=['invalid'])]},
|
||||
families=['iclr'], years=[2017])
|
||||
self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
|
||||
self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
|
||||
|
||||
def workshop_cases(self):
|
||||
return [
|
||||
(2024, 'workshop_me_fomo', 'ICLR 2024 Workshop MEFoMo Withdrawn Submission', 'withdrawn', 'ME-FoMo'),
|
||||
(2024, 'workshop_genai4dm', 'ICLR 2024 Workshop GenAI4DM Withdrawn Submission', 'withdrawn', 'GenAI4DM'),
|
||||
(2025, 'workshop_agenticai', 'ICLR 2025 Workshop AgenticAI Reject', 'rejected', 'AgenticAI'),
|
||||
(2025, 'workshop_lmrl', 'ICLR 2025 Workshop LMRL Withdrawn Submission', 'withdrawn', 'LMRL'),
|
||||
(2026, 'workshop_llm_reasoning', 'ICLR 2026 Workshop LLM Reasoning Withdrawn Submission', 'withdrawn', 'LLM_Reasoning'),
|
||||
]
|
||||
|
||||
def test_iclr_workshop_labels_require_opt_in_and_retain_identity_evidence(self):
|
||||
for year, track, label, status, group in self.workshop_cases():
|
||||
with self.subTest(label=label):
|
||||
raw = self.row(track=track, status=label)
|
||||
self.assertEqual(normalize(raw, 'iclr', year, self.provenance()),
|
||||
(None, 'separate_workshop'))
|
||||
row, reason = normalize(raw, 'iclr', year, self.provenance(), True)
|
||||
self.assertIsNone(reason)
|
||||
self.assertEqual((row['venue'], row['year'], row['track'], row['status']),
|
||||
('ICLR', year, track, status))
|
||||
self.assertEqual(row['raw_decision'], label)
|
||||
self.assertEqual(row['metadata']['workshop_venueid'],
|
||||
f'ICLR.cc/{year}/Workshop/{group}')
|
||||
self.assertFalse(row['metadata']['original_decision_signature_verified'])
|
||||
self.assertEqual(row['evidence_url'], 'https://example.org/pinned.jsonl#L7')
|
||||
self.assertNotIn('private review', json.dumps(row))
|
||||
|
||||
def test_workshop_label_scope_conflicts_never_become_main_events(self):
|
||||
for year, track, label, _, _ in self.workshop_cases():
|
||||
for enabled in (False, True):
|
||||
for wrong_track in ('main', 'workshop_other'):
|
||||
self.assertEqual(normalize(self.row(track=wrong_track, status=label),
|
||||
'iclr', year, self.provenance(), enabled),
|
||||
(None, 'conflicting_workshop_track'))
|
||||
for family, wrong_year in [('icml', year), ('iclr', year + 1)]:
|
||||
with self.assertRaisesRegex(ValueError, 'conference year'):
|
||||
normalize(self.row(track=track, status=label), family,
|
||||
wrong_year, self.provenance(), enabled)
|
||||
row, reason = normalize(self.row(track='workshop_unknown',
|
||||
status='ICLR 2026 Workshop Unknown Reject'),
|
||||
'iclr', 2026, self.provenance(), True)
|
||||
self.assertIsNone(row)
|
||||
self.assertEqual(reason, 'unmapped_negative_label')
|
||||
|
||||
def test_workshop_collection_and_import_keep_five_tracks_separate_from_main(self):
|
||||
files = {}
|
||||
for index, (year, track, label, _, _) in enumerate(self.workshop_cases()):
|
||||
files.setdefault(f'venues/iclr/iclr{year}.jsonl', []).append(
|
||||
self.row(id=f'Workshop{index:02}', track=track, status=label))
|
||||
files['venues/iclr/iclr2024.jsonl'].append(self.row())
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
default = self.collect(root / 'default', files, families=['iclr'])
|
||||
self.assertEqual(default['normalized_records'], 1)
|
||||
held = [json.loads(line) for line in Path(default['held_file']).read_bytes().split(b'\n') if line]
|
||||
self.assertEqual(len(held), 5)
|
||||
self.assertTrue(all(row['reason'] == 'separate_workshop' for row in held))
|
||||
selected = self.collect(root / 'selected', files, families=['iclr'], include_workshops=True)
|
||||
self.assertEqual(selected['normalized_records'], 6)
|
||||
self.assertEqual(len(selected['files']), 6)
|
||||
self.assertEqual(Path(selected['held_file']).read_bytes(), b'')
|
||||
db = root / 'db.sqlite'
|
||||
for manifest in selected['files']:
|
||||
self.assertEqual(manifest['parser_version'], 3)
|
||||
ingest_file(db, manifest['file'])
|
||||
store = Store(db)
|
||||
try:
|
||||
scopes = set(store.conn.execute('SELECT e.year,e.track,d.status FROM submissions s '
|
||||
'JOIN editions e ON e.id=s.edition_id '
|
||||
'JOIN decisions d ON d.submission_id=s.id'))
|
||||
expected = {(year, track, status) for year, track, _, status, _ in self.workshop_cases()}
|
||||
self.assertEqual(scopes, expected | {(2024, 'main', 'rejected')})
|
||||
self.assertEqual(store.conn.execute('PRAGMA foreign_key_check').fetchall(), [])
|
||||
finally:
|
||||
store.conn.close()
|
||||
|
||||
def test_mirror_reimport_preserves_one_attempt_and_later_acceptance_elsewhere(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
@@ -152,6 +269,35 @@ class PublicOpenReviewTests(unittest.TestCase):
|
||||
finally:
|
||||
store.conn.close()
|
||||
|
||||
def test_unicode_separators_in_abstract_survive_audit_and_import(self):
|
||||
for separator in ('\u0085', '\u2028', '\u2029'):
|
||||
with self.subTest(separator=repr(separator)), tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
abstract = 'First paragraph' + separator + 'Second paragraph'
|
||||
result = self.collect(root / 'source', {
|
||||
'venues/icml/icml2025.jsonl': [self.row(abstract=abstract)]})
|
||||
path = Path(result['files'][0]['file'])
|
||||
data = path.read_bytes()
|
||||
self.assertIn(separator.encode(), data)
|
||||
self.assertEqual(data.count(b'\n'), 1)
|
||||
db = root / 'db.sqlite'
|
||||
store = Store(db)
|
||||
store.conn.close()
|
||||
inventory = root / 'source/public-openreview-inventory.json'
|
||||
before = audit(db, inventory, root / 'before.json')
|
||||
self.assertEqual(before['status'], 'ready_for_serial_import')
|
||||
self.assertEqual(before['records_checked'], 1)
|
||||
ingest_file(db, path)
|
||||
after = audit(db, inventory, root / 'after.json')
|
||||
self.assertEqual(after['status'], 'ready_for_serial_import')
|
||||
self.assertEqual(after['existing_same_edition_attempts'], 1)
|
||||
store = Store(db)
|
||||
try:
|
||||
self.assertEqual(store.conn.execute(
|
||||
'SELECT abstract FROM source_records').fetchone()[0], abstract)
|
||||
finally:
|
||||
store.conn.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
"""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.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
|
||||
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_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'))
|
||||
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')
|
||||
@@ -133,7 +133,7 @@ class SubmissionTests(unittest.TestCase):
|
||||
c.execute("INSERT INTO papers(id,title,title_key) VALUES(1,'Paper','paper')")
|
||||
c.execute("INSERT INTO source_records VALUES(1,1,'paper',1,'https://example.org/p','Paper','Abstract','[]','{}',CURRENT_TIMESTAMP)")
|
||||
c.execute("INSERT INTO venues VALUES(1,'ICLR','ICLR','machine_learning','conference')")
|
||||
c.execute("INSERT INTO editions VALUES(1,1,2026,'main')")
|
||||
c.execute("INSERT INTO editions(id,venue_id,year,track) VALUES(1,1,2026,'main')")
|
||||
c.execute("INSERT INTO submissions VALUES(1,1,1,1,'paper',NULL)")
|
||||
c.execute("INSERT INTO decisions VALUES(1,1,'accepted','Accept','https://example.org/decision','2026-01-01T00:00:00Z',NULL)")
|
||||
c.commit();c.close()
|
||||
|
||||
@@ -37,6 +37,27 @@ class VisionTests(unittest.TestCase):
|
||||
collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
|
||||
self.assertFalse(list(Path(tmp).glob('*.jsonl')))
|
||||
|
||||
def test_fragment_variants_share_one_all_papers_resource(self):
|
||||
root,papers,entry=self.cvf_fixture();all_url=root+'?day=all'
|
||||
pages={root:''.join(f'<a href="{all_url}{fragment}">All</a>' for fragment in ('','#','#papers')),
|
||||
all_url:'<dl>'+entry(papers[0])+entry(papers[1])+'</dl>'}
|
||||
pages.update({paper:'<div id="papertitle">Paper</div><div id="abstract">Abstract</div>' for paper in papers[:2]})
|
||||
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=lambda url,*a,**kw:pages[url].encode()) as fetch:
|
||||
manifest=collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
|
||||
self.assertEqual(manifest['count'],2)
|
||||
self.assertEqual(manifest['index_urls'],[all_url])
|
||||
self.assertEqual([call.args[0] for call in fetch.call_args_list].count(all_url),1)
|
||||
self.assertFalse(any('#' in call.args[0] for call in fetch.call_args_list))
|
||||
|
||||
def test_distinct_all_papers_queries_still_fail_before_publishing(self):
|
||||
root,_,_=self.cvf_fixture();urls=[root+'?day=all',root+'?day=all&view=alternate']
|
||||
html=''.join(f'<a href="{url}">All</a>' for url in urls).encode()
|
||||
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',return_value=html) as fetch:
|
||||
with self.assertRaisesRegex(ValueError,'Ambiguous all-papers index'):
|
||||
collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
|
||||
self.assertEqual(fetch.call_count,1)
|
||||
self.assertFalse(list(Path(tmp).glob('*.jsonl')))
|
||||
|
||||
def test_missing_abstract_page_keeps_official_index_metadata_and_marks_partial(self):
|
||||
root,papers,entry=self.cvf_fixture();error=HTTPError(papers[0],404,'Not Found',{},io.BytesIO())
|
||||
def fetch(url,*args,**kwargs):
|
||||
|
||||
Reference in New Issue
Block a user