Recover omitted DBLP discovery families from pinned snapshots

.gitea/workflows/check.yml

@@ -12,7 +12,7 @@ jobs:
       - name: Verify metadata storage and query commands
         run: |
           python3 -B - <<'PY'
-          import gzip, io, json, os, pathlib, re, subprocess, tarfile, tempfile, urllib.request
+          import io, os, pathlib, re, subprocess, tarfile, tempfile, urllib.request
           sha = os.environ['ARCHIVE_SHA']
           if not re.fullmatch(r'[0-9a-fA-F]{40,64}', sha):
               raise SystemExit('Invalid source commit')
@@ -50,21 +50,10 @@ jobs:
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','www-archive','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','aamas-proceedings','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','dblp-family','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','aamas-proceedings','--year','2025','--snapshots',str(src/'tests/fixtures/aamas2025'),'--out',str(root/'installed-aamas2025')],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','iclr-archive','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','acl','--help'],cwd=outside,check=True)
-              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','springer-selected','--help'],cwd=outside,check=True)
-              springer = json.loads(gzip.decompress((src/'tests/fixtures/springer-aistats1993-fixture.json.gz').read_bytes()))
-              for group in ('toc','chapters'):
-                  folder = root/'springer-snapshots'/group
-                  folder.mkdir(parents=True)
-                  for name, content in springer[group].items():
-                      (folder/name).write_text(content)
-              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','springer-selected','--toc-snapshots',str(root/'springer-snapshots/toc'),'--chapter-snapshots',str(root/'springer-snapshots/chapters'),'--out',str(root/'installed-springer')],cwd=outside,check=True)
-              springer_report = json.loads((root/'installed-springer/collection-report.json').read_bytes())
-              assert springer_report['records'] == 49 and not springer_report['database_written']
-              springer_manifest, = springer_report['files']
-              assert springer_manifest['status'] == 'partial' and springer_manifest['missing_abstract_count'] == 0
               subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','edition-tracks','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','joint-editions','--help'],cwd=outside,check=True)

README.md

@@ -8,8 +8,6 @@ Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追
 
 历史 ICLR 2014–2016 官方录用页由 `collect iclr-archive` 适配，保留主会/workshop及重复展示证据；运行和身份核验边界见 [历史档案说明](docs/coverage/iclr-archive-2014-2016.md)。
 
-`collect springer-selected` 核验限定的会后选集，分开保存会议年份、出版年份及选集事件身份。目前支持 AISTATS 1993 对应的 1994 年出版选集，原始会议决定仍须单独核查；配置、证据和恢复见 [会后选集说明](docs/springer-selected-volumes.md)。
-
 ## 项目结构
 
 ACL 1978 的年会与 TINLAP-2 来源关系按精确卷适配，保留原系列和不明逐篇角色；1965 的替代关系不转成全局别名。来源、排除条目和恢复边界见 [ACL 早期来源核查](docs/coverage/acl-early-sources-2026-11-09.md)。
@@ -72,6 +70,8 @@ EWSL、EPS、ECP、EWSP 的[具体前身卷与完整快照要求](docs/coverage/
 
 AAMAS 的[官方论文集采集器](docs/aamas-proceedings.md)支持已核验的 2012、2025 版式，可从缓存或已校验快照采集元数据。2025 年目录含 476 条非 keynote 出版物，保留完整作者和机构，HTML 未提供摘要时留空；发表形式与投稿 track 分开，不根据主题推定主会轨道或研究类型。
 
+旧 DBLP 候选文件可能漏掉新命名空间。`collect dblp-family` 可从已校验的固定压缩快照和完整目录中[恢复指定 family 的原始候选](docs/dblp-family-recovery.md)，例如 `ifaamas`；恢复文件保留卷、keynote 和 workshop 等原始条目，需另做来源范围与身份核验。
+
 `paper-atlas collect --help` 查看来源；`paper-atlas import --help` 查看导入方式。完整批次步骤见[使用指南](docs/usage.md)。
 
 共享实例名为 `foundry-postgres`，数据位于 `foundry_data` 库的 `research` schema；其他项目可按权限复用同一组表。库名通过 libpq service 配置、schema 通过 search_path 或 `--pg-schema` 指定，不写死在查询中。PostgreSQL 凭据放在服务器权限为 `0600` 的 libpq service 文件中；命令行只使用服务名称。`paper-atlas maintain postgres --help` 提供迁移入口：先完整核验 21 张业务表，再发布目标 schema。SQLite FTS5 内部表由 PostgreSQL GIN 索引替代。切换后 SQLite 为只读备份；`import records`、`import arxiv` 和 `maintain backfill` 使用 `--pg-service foundry-research` 写入主库，整个写入会话只允许一个协调写入者，会议文件失败时整体回滚并保留失败批次记录。

docs/dblp-family-recovery.md

@@ -0,0 +1,37 @@
+# DBLP family 候选恢复
+
+DBLP 的会议命名空间会变化。旧候选过滤器包含 `atal`、`aamas`，遗漏了 `ifaamas`；2026-11-01 固定快照的完整目录实际已有 `conf/ifaamas/2025` 和 `conf/ifaamas/2026`。新目录盘点已把 `ifaamas` 纳入广泛候选范围，已有快照也可按需恢复，无需重新下载或重建完整目录。
+
+## 运行
+
+在已有元数据的服务器运行：
+
+```sh
+paper-atlas collect dblp-family \
+  --archive /data/paper-index/raw/dblp-2026-11-01.xml.gz \
+  --dtd /data/paper-index/raw/dblp-2023-06-28.dtd \
+  --catalogue /data/paper-index/collected/dblp/dblp-catalogue.sqlite \
+  --family ifaamas \
+  --out /data/paper-index/collected/ifaamas-recovered
+```
+
+`--family` 可以重复，但仅接受项目明确列出的候选 family。`--out` 必须尚不存在。同名输出由文件锁防止并发恢复；任务运行中应观测原任务，不能因为日志间隔长就重复启动。完整扫描会定期打印进度。
+
+命令重新验证固定压缩包 SHA256、DTD MD5 与大小，并只读打开目录。目录旁的 `dblp-ai-candidates.manifest.json` 必须属于同一个快照和 DTD，包含完整记录类型计数。旧候选文件不作为完整性依据，也不会被改写。
+
+恢复扫描使用项目原有 XML 流式解析器，仅允许已校验的本地 DTD，不联网读取 XML 实体。最终会比对完整快照的记录总数、逐 family 数量和全部选中卷的原始元数据。全部一致后才原子发布输出目录；失败时不发布半成品，也不覆盖既有文件。排查明确终止的失败任务后，可用尚不存在的输出目录重新运行。
+
+## 输出与后续流程
+
+- `dblp-family-candidates.jsonl`：原始 `inproceedings` 和 `proceedings`，保留书目 key、原始作者、标题、年份、卷引用、出版链接等字段。
+- `recovery-manifest.json`：快照、DTD、目录凭据哈希，完整扫描数量、逐 family 数量、选中卷、旧候选文件的已知数量及输出哈希。状态为 `recovered_raw_candidates_pending_scope_review`。
+
+恢复结果没有新增规范化的 `venue`、`track` 或决定标签，也不是 `import records` 的输入队列。`ifaamas` 的整卷可能混有研究论文、扩展摘要、keynote、博士生、演示和 JAAMAS 展示。目录过滤修复不会把这些条目统一标成 main；规范化的范围映射仍需第一方逐栏目证据。
+
+后续先用官方目录核对标题、页码、作者及发表形式，再查数据库稳定标识和候选实体，保留每次投稿、决定和标签来源。只有审核后的新增分区才能原生导入并独立验收；已验收的来源记录不能重放。单个年份有记录不代表逐篇、track、摘要或公开拒稿覆盖完整。
+
+## 2026-11-09 核查实例
+
+对既有固定快照的独立原始元数据恢复发现 `ifaamas` 有 1,122 条出版物候选及 2 卷：2025 年 479 条、2026 年 643 条。2025 年的 479 条标题与起始页均对应官方目录，其中 3 条 keynote 应排除，余下 476 条出版物需逐项处理。该数字包含已经验收的 225 条来源，不是 1,122 篇可新增论文。
+
+此实例说明过滤遗漏的影响，不代表恢复命令已完成生产导入。2026 年仍需核验官方出版清单、栏目、身份及逐次决定；缺少录用记录不能推断为拒稿。最终全库去重、覆盖和查询审计继续单独进行。

docs/springer-selected-volumes.md

@@ -1,36 +0,0 @@
-# 会后选集的元数据采集
-
-`collect springer-selected` 读取已冻结、带 HTTP 回执的公开 HTML，不访问 PDF、不写业务库。目前仅支持核验过的 *Selecting Models from Data: Artificial Intelligence and Statistics IV*，书籍 DOI 为 `10.1007/978-1-4612-2660-4`。增加其他卷前必须另行核对会议关系、目录、年份和字段格式。
-
-[会议官方旧站](https://www.gatsby.ucl.ac.uk/aistats/workshops.html)区分现场分发的 Preliminary Papers 与之后修订、再次评审的选集。[出版社卷说明](https://link.springer.com/book/10.1007/978-1-4612-2660-4)将本书关联到 1993 年 1 月的第四届 workshop，书籍出版年份为 1994。三页目录共有 49 个章节 DOI；分区 Front Matter 不作为章节论文导入。
-
-规范记录保留：
-
-- 会议年份 1993、独立 track `selected_volume` 与事件标识 `aistats-1993-selected-volume-1994`；出版年份单独记为 1994。
-- 章节本身的 DOI、作者原始引用写法及顺序、姓名首字母、标题、摘要、页码和逐页原始证据。不扩写作者首字母。
-- 明确提供本章 `publication_doi` 投稿出版标识，不把它转移到旧的原始会议评审记录，也不因短标题相同而强制合并实体。
-- `springer_selected_volume:publication_format` 标签与独立来源 namespace。
-- `accepted` 仅表示章节已列入这本出版选集，原始 1993 投稿评审决定、决定日期和全部录用清单均未知。后续原会议 main、拒稿或撤稿证据可以独立保留。
-
-不能据此宣称会议全部录用、全部投稿或公开拒稿覆盖完整，也不能把出版社机器添加的关键词视为作者标签。未命中选集不等于被拒。
-
-## 按需运行与恢复
-
-输入分别是卷目录/旧站说明和章节页的快照目录。每个目录的 `receipts.json` 包含 `responses` 列表，每条保留 `url`、`final_url`、`file`、`http_status`、`headers`、`bytes`、`sha256`、`at`。需要三页目录、官方旧站说明及全部 49 个章节页面。
-
-```sh
-paper-atlas collect springer-selected \
-  --toc-snapshots /data/paper-index/collected/aistats1993-book-evidence-v1 \
-  --chapter-snapshots /data/paper-index/collected/aistats1993-chapter-evidence-v1 \
-  --out /data/paper-index/collected/aistats1993-source-v1
-```
-
-采集器验证回执和内容哈希、文件路径、HTTP 类型、目录/章节 DOI 和标题、作者格式、出版年份及页码。Springer 对公开元数据返回的同路径 `cookies_not_supported` 诊断参数会原样保留；不接受其他域、路径、分页变化或未知查询参数。官方旧站说明固定哈希为 `6db205b9dbc82b4a096556efd2aa67e065cad8e0bd6351176b1962ffd1fca179`，变化时须重新审核。
-
-全部文件先核验，输出使用排他锁和暂存目录，完成报告最后发布。已有输出拒绝覆盖；发现不完整或已消费文件，应先查运行状态与报告，另建明确版本目录，不能重放生产导入。
-
-输出 JSONL 配套 manifest 始终为 `partial`。收集完成后还必须做 PostgreSQL 稳定标识/标题/作者身份预检、排除已导入记录、准备限定增量文件、保存旧数据保护基线、原生导入和独立验收。49 条来源不代表 49 个新增论文实体，全部来源处理后仍须全库去重审计。
-
-## 固定测试资料
-
-`tests/fixtures/springer-aistats1993-fixture.json.gz` 保存 4 份范围/目录 HTML、49 份章节 HTML、原始请求回执及独立原始字段审计的预期元数据，SHA256 为 `94028fd37080f9d8027e5790e379cc151ee6eab17ed0d1dcc0f59e6e75f9d4d0`。摘要按 HTML 段落边界分隔；未修改原始 HTML 或回执。它是压缩的公开元数据测试资料，不包含论文 PDF。测试覆盖全部真实章节字段、缺失和漂移输入、来源安全边界、幂等导入及旧拒稿历史保留。构建和测试只在已授权远程 runner 执行。

paper_atlas/cli.py

@@ -13,7 +13,6 @@ COMMANDS = {
     ('download',): ('paper_atlas.io.snapshots', {}, '下载并校验固定版本快照'),
     ('collect','pmlr'): ('paper_atlas.collectors.pmlr', {}, 'PMLR 官方元数据'),
     ('collect','acl'): ('paper_atlas.collectors.acl', {}, 'ACL Anthology XML'),
-    ('collect','springer-selected'): ('paper_atlas.collectors.springer_selected', {}, '核验限定的会后选集章节与原会议关系'),
     ('collect','iclr-archive'): ('paper_atlas.collectors.iclr_archive', {}, '核对ICLR 2014–2016官方历史录用档案'),
     ('collect','iclr-program'): ('paper_atlas.collectors.iclr', {}, 'ICLR 官方公开程序'),
     ('collect','iclr-snapshot'): ('paper_atlas.collectors.openreview', {}, 'ICLR 公开投稿与决定快照'),
@@ -28,6 +27,7 @@ COMMANDS = {
     ('collect','aamas-proceedings'): ('paper_atlas.collectors.aamas', {}, '按官方目录区分 AAMAS 发表形式和摘要'),
     ('collect','www-archive'): ('paper_atlas.collectors.www_archive', {}, '核验早期WWW官方档案并分开待核查条目'),
     ('collect','dblp'): ('paper_atlas.collectors.dblp_records', {}, '核对并整理 DBLP 历史发表记录'),
+    ('collect','dblp-family'): ('paper_atlas.collectors.dblp_recovery', {}, '从固定快照恢复指定 DBLP family 的原始候选'),
     ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
     ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
     ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),

paper_atlas/collectors/dblp.py

@@ -20,7 +20,7 @@ DTD_MD5='96e4283e71886dec2173718957293648'
 DTD_NAMES={'dblp.dtd','dblp-2023-06-28.dtd'}
 RELEASE='https://doi.org/10.4230/dblp.xml.2026-11-01'
 # Broad discovery candidates, including historical names; not a venue mapping.
-FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps ecp ewsp kr atal aamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())
+FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps ecp ewsp kr atal aamas ifaamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())
 
 
 def records(archive,dtd_bytes):

paper_atlas/collectors/dblp_recovery.py

@@ -0,0 +1,156 @@
+"""Recover raw family candidates from a pinned snapshot and its existing catalogue.
+
+The old candidate file is not an exhaustive source: earlier discovery filters
+can omit an entire namespace. This command checks the complete snapshot and
+catalogue, without assigning venues, tracks, acceptance or paper identities.
+"""
+import argparse
+from collections import Counter
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import re
+import sqlite3
+import tempfile
+
+from paper_atlas.collectors.dblp import DTD_MD5, FAMILIES, RELEASE, XML_SHA, family, records
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.snapshots import digest_file
+
+
+def catalogue_evidence(catalogue, families, expected_sha, expected_dtd_md5):
+    manifest_path = catalogue.with_name('dblp-ai-candidates.manifest.json')
+    manifest = json.loads(manifest_path.read_text())
+    if (manifest.get('snapshot_sha256') != expected_sha
+            or manifest.get('dtd_md5') != expected_dtd_md5):
+        raise ValueError('Catalogue belongs to a different snapshot or DTD')
+    totals = manifest.get('record_types', {})
+    if (set(totals) != {'proceedings', 'inproceedings'}
+            or any(type(n) is not int or n < 0 for n in totals.values())
+            or type(manifest.get('records_scanned')) is not int
+            or sum(totals.values()) != manifest['records_scanned']):
+        raise ValueError('Incomplete catalogue record totals')
+    counts = Counter()
+    volumes = {}
+    # The catalogue is evidence, never a writable recovery destination.
+    with closing(sqlite3.connect(catalogue.resolve().as_uri() + '?mode=ro', uri=True)) as c:
+        for name in families:
+            found = c.execute('SELECT record_type,count FROM family_counts WHERE family=?',
+                              (name,)).fetchall()
+            if not found or not any(n for _, n in found):
+                raise ValueError('Family absent from complete catalogue: ' + name)
+            for kind, count in found:
+                if kind not in totals or type(count) is not int or count < 0:
+                    raise ValueError('Invalid catalogue family count')
+                counts[(name, kind)] = count
+            for key, raw in c.execute('SELECT key,metadata_json FROM proceedings WHERE family=?',
+                                      (name,)):
+                row = json.loads(raw)
+                if (family(key) != name or row.get('key') != key
+                        or row.get('record_type') != 'proceedings'):
+                    raise ValueError('Invalid catalogue proceedings evidence')
+                volumes[key] = row
+    if len(volumes) != sum(counts[(name, 'proceedings')] for name in families):
+        raise ValueError('Catalogue proceedings count differs from volume records')
+    return manifest, digest_file(manifest_path)['sha256'], counts, volumes
+
+
+def recover(archive, dtd, catalogue, out, families, *, expected_sha=XML_SHA,
+            expected_dtd_md5=DTD_MD5):
+    families = list(families)
+    if (not families or any(not isinstance(name, str)
+                           or not re.fullmatch(r'[a-z][a-z0-9-]*', name)
+                           or name not in FAMILIES for name in families)):
+        raise ValueError('Select an explicitly supported discovery family')
+    families = sorted(set(families))
+    archive, dtd, catalogue = Path(archive), Path(dtd), Path(catalogue)
+    output = Path(out).absolute()
+    if output.exists() or output.is_symlink():
+        raise ValueError('Recovery output must not already exist')
+    output.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = output.parent / ('.' + output.name + '.recovery.lock')
+    with lock_path.open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if output.exists() or output.is_symlink():
+            raise ValueError('Recovery output appeared before scan')
+        actual = digest_file(archive)
+        if actual['sha256'] != expected_sha:
+            raise ValueError('DBLP snapshot checksum mismatch')
+        if dtd.stat().st_size > 128 * 1024 or digest_file(dtd)['md5'] != expected_dtd_md5:
+            raise ValueError('DBLP DTD checksum mismatch or size limit')
+        manifest, manifest_sha, expected, volumes = catalogue_evidence(
+            catalogue, families, expected_sha, expected_dtd_md5)
+        totals, found = Counter(), Counter()
+        seen_keys, found_volumes = set(), set()
+        with tempfile.TemporaryDirectory(prefix='.dblp-family-', dir=output.parent) as tmp:
+            stage = Path(tmp) / 'recovered'
+            stage.mkdir()
+            target = stage / 'dblp-family-candidates.jsonl'
+            with target.open('w') as stream:
+                for row in records(archive, dtd.read_bytes()):
+                    totals[row['record_type']] += 1
+                    name = family(row['key'])
+                    if name in families:
+                        if row['key'] in seen_keys:
+                            raise ValueError('Duplicate raw bibliographic key')
+                        seen_keys.add(row['key'])
+                        found[(name, row['record_type'])] += 1
+                        if row['record_type'] == 'proceedings':
+                            if volumes.get(row['key']) != row:
+                                raise ValueError('Raw proceedings differs from catalogue evidence')
+                            found_volumes.add(row['key'])
+                        stream.write(json.dumps(row, ensure_ascii=False) + '\n')
+                    scanned = sum(totals.values())
+                    if scanned % 250000 == 0:
+                        print(json.dumps({'status': 'scanning_raw_family_candidates',
+                                          'records_scanned': scanned,
+                                          'recovered_records': sum(found.values())}), flush=True)
+            if (totals != Counter(manifest['record_types'])
+                    or sum(totals.values()) != manifest['records_scanned']):
+                raise ValueError('Complete snapshot totals differ from catalogue')
+            if found != expected or found_volumes != volumes.keys():
+                raise ValueError('Recovered family totals differ from catalogue')
+            report = {
+                'format_version': 1, 'source': 'dblp',
+                'status': 'recovered_raw_candidates_pending_scope_review',
+                'finished_at': timestamp(), 'source_release': RELEASE,
+                'snapshot_sha256': actual['sha256'], 'dtd_md5': expected_dtd_md5,
+                'catalogue_manifest_sha256': manifest_sha, 'families': families,
+                'records_scanned': sum(totals.values()), 'record_types': dict(totals),
+                'recovered_records': sum(found.values()),
+                'recovered_publications': sum(found[(n, 'inproceedings')] for n in families),
+                'recovered_volumes': len(found_volumes),
+                'family_counts': [{'family': name, 'record_type': kind, 'count': count}
+                                  for (name, kind), count in sorted(found.items())],
+                'old_candidate_family_counts': {
+                    name: manifest.get('candidate_family_counts', {}).get(name, 0)
+                    for name in families},
+                'catalogue_volumes': [volumes[k] for k in sorted(volumes)],
+                'file': str(output / target.name),
+                'file_sha256': digest_file(target)['sha256'], 'database_written': False,
+                'note': 'Raw bibliographic candidates, including volume records and possible '
+                        'keynotes/workshops. No venue, track, acceptance or work identity '
+                        'inferred. Requires primary-source scope and identity review; '
+                        'not a normalized import queue. Existing imported sources must '
+                        'not be replayed.',
+            }
+            # Deliberately not an import-records sidecar manifest.
+            (stage / 'recovery-manifest.json').write_text(
+                json.dumps(report, ensure_ascii=False, indent=2) + '\n')
+            if output.exists() or output.is_symlink():
+                raise ValueError('Recovery output appeared during scan')
+            stage.rename(output)
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog, description=__doc__)
+    parser.add_argument('--archive', required=True, type=Path)
+    parser.add_argument('--dtd', required=True, type=Path)
+    parser.add_argument('--catalogue', required=True, type=Path)
+    parser.add_argument('--family', required=True, action='append', choices=sorted(FAMILIES))
+    parser.add_argument('--out', required=True, type=Path)
+    args = parser.parse_args(argv)
+    print(json.dumps(recover(args.archive, args.dtd, args.catalogue, args.out, args.family),
+                     ensure_ascii=False))

paper_atlas/collectors/springer_selected.py

@@ -1,231 +0,0 @@
-"""Reviewed postconference selected volumes from frozen public HTML metadata."""
-import argparse
-from collections import defaultdict
-import fcntl
-import hashlib
-import json
-from pathlib import Path
-import re
-import tempfile
-from urllib.parse import parse_qs, urlsplit
-
-from paper_atlas.db.submissions import timestamp
-from paper_atlas.io.html import Document, clean
-
-SOURCE = 'springer_selected_volume'
-BOOK = '10.1007/978-1-4612-2660-4'
-BOOK_URL = 'https://link.springer.com/book/'+BOOK
-HISTORY_URL = 'https://www.gatsby.ucl.ac.uk/aistats/workshops.html'
-HISTORY_SHA = '6db205b9dbc82b4a096556efd2aa67e065cad8e0bd6351176b1962ffd1fca179'
-SCOPE = {
-    'venue': 'AISTATS', 'year': 1993, 'track': 'selected_volume',
-    'event_key': 'aistats-1993-selected-volume-1994',
-    'publication_year': 1994, 'book_doi': BOOK,
-    'book_title': 'Selecting Models from Data', 'expected_chapters': 49,
-    'relationship': 'Selected workshop papers were revised and rereviewed for later publication; this is not the original complete workshop decision list.',
-}
-
-
-def snapshot_index(root):
-    root = Path(root).resolve()
-    receipts = json.loads((root/'receipts.json').read_bytes())['responses']
-    index = {}
-    for receipt in receipts:
-        url = receipt['url']
-        if url in index:
-            raise ValueError('Duplicate snapshot URL')
-        index[url] = (root, receipt)
-    return index
-
-
-def read_page(index, url):
-    if url not in index:
-        raise ValueError('Missing required public HTML snapshot: '+url)
-    root, receipt = index[url]
-    filename = Path(receipt['file'])
-    path = root/filename
-    content_type = {k.lower():v for k,v in receipt.get('headers', {}).items()}.get('content-type', '')
-    requested, final = urlsplit(url), urlsplit(receipt.get('final_url', ''))
-    query = parse_qs(final.query, keep_blank_values=True)
-    original_query = parse_qs(requested.query, keep_blank_values=True)
-    # Springer returns the public metadata with an explicit cookie diagnostic.
-    # Retain that final URL verbatim; accept no other redirect or changed page.
-    if query != original_query and requested.netloc == 'link.springer.com':
-        if (query.pop('error', None) != ['cookies_not_supported']
-                or not re.fullmatch(r'[0-9a-f-]{36}', ''.join(query.pop('code', [])))):
-            raise ValueError('Unreviewed Springer redirect query')
-    if (filename.is_absolute() or '..' in filename.parts or not path.resolve().is_relative_to(root)
-            or receipt.get('http_status') != 200
-            or (final.scheme, final.netloc, final.path) != (requested.scheme, requested.netloc, requested.path)
-            or final.fragment or query != original_query
-            or 'text/html' not in content_type.lower()):
-        raise ValueError('Unsafe, redirected or unsuccessful public HTML snapshot')
-    raw = path.read_bytes()
-    digest = hashlib.sha256(raw).hexdigest()
-    if len(raw) > 2_000_000 or len(raw) != receipt['bytes'] or digest != receipt['sha256']:
-        raise ValueError('Snapshot checksum or size differs')
-    evidence = {'url':url, 'sha256':digest, 'bytes':len(raw),
-                'file':'evidence/'+digest+'.html', 'http_status':200,
-                'observed_at':timestamp(receipt['at']), 'response_receipt':receipt}
-    return raw, evidence
-
-
-def parse_toc(html, *, excluded_materials=None):
-    entries = {}
-    for node in Document(html).root.walk():
-        if node.tag != 'li' or node.attrs.get('data-test') != 'chapter':
-            continue
-        links = [n for n in node.walk() if n.tag == 'a' and re.fullmatch(
-            r'/chapter/'+re.escape(BOOK)+r'_[1-9]\d*', n.attrs.get('href', ''))]
-        if not links:
-            headings = [clean(n.text()) for n in node.walk() if n.tag in ('h3','h4')]
-            frontmatter = [n.attrs.get('href','') for n in node.walk()
-                           if n.tag == 'a' and n.attrs.get('data-test') == 'front-matter-pdf']
-            if headings == ['Front Matter'] and len(frontmatter) == 1:
-                if excluded_materials is not None:
-                    excluded_materials.append({'title':'Front Matter', 'url':frontmatter[0],
-                                               'reason':'Section front matter without a chapter DOI; not an authored chapter.'})
-                continue
-        if len(links) != 1:
-            raise ValueError('Chapter entry lacks one reviewed-book DOI')
-        link, = links
-        doi = link.attrs['href'].removeprefix('/chapter/')
-        if doi in entries or not clean(link.text()):
-            raise ValueError('Duplicate DOI or missing chapter title')
-        entries[doi] = clean(link.text())
-    return entries
-
-
-def parse_chapter(html, doi):
-    nodes = list(Document(html).root.walk())
-    meta = defaultdict(list)
-    for node in nodes:
-        if node.tag == 'meta' and node.attrs.get('name', '').startswith('citation_'):
-            meta[node.attrs['name']].append(node.attrs.get('content', ''))
-    def one(key):
-        values = meta[key]
-        if len(values) != 1 or not values[0]:
-            raise ValueError('Missing or ambiguous citation field: '+key)
-        return values[0]
-    if (one('citation_doi') != doi or one('citation_inbook_title') != SCOPE['book_title']
-            or one('citation_publication_date') != str(SCOPE['publication_year'])):
-        raise ValueError('Chapter DOI, book or publication year differs from reviewed scope')
-    raw_authors = meta['citation_author']
-    if not raw_authors:
-        raise ValueError('Missing named chapter authors')
-    authors = []
-    for raw in raw_authors:
-        if raw.count(',') != 1:
-            raise ValueError('Unreviewed citation author format')
-        family, given = [clean(s) for s in raw.split(',', 1)]
-        if not family or not given:
-            raise ValueError('Incomplete citation author')
-        authors.append({'given':given, 'family':family})
-    abstract_nodes = [n for n in nodes if n.attrs.get('id') == 'Abs1-content']
-    if len(abstract_nodes) > 1:
-        raise ValueError('Ambiguous abstract section')
-    abstract = clean(abstract_nodes[0].text()) if abstract_nodes else None
-    first, last = one('citation_firstpage'), one('citation_lastpage')
-    if not first.isdigit() or not last.isdigit() or int(first) > int(last):
-        raise ValueError('Invalid chapter pages')
-    return {'doi':doi, 'title':one('citation_title'), 'authors':authors,
-            'authors_citation_raw':list(raw_authors), 'abstract':abstract or None,
-            'first_page':first, 'last_page':last}
-
-
-def record_for(chapter, evidence, scope_evidence):
-    return {
-        'source':SOURCE, 'external_id':chapter['doi'],
-        'url':'https://link.springer.com/chapter/'+chapter['doi'],
-        'title':chapter['title'], 'abstract':chapter['abstract'], 'authors':chapter['authors'],
-        'identifiers':{'doi':chapter['doi']},
-        'submission_identifiers':[{'scheme':'publication_doi','value':chapter['doi']}],
-        **{k:SCOPE[k] for k in ('venue','year','track','event_key')},
-        'status':'accepted', 'observed_at':evidence['observed_at'],
-        'raw_decision':'listed in the 1994 selected postconference volume; original 1993 review decision not supplied',
-        'tags':[{'namespace':SOURCE+':publication_format','value':'selected_volume_chapter'}],
-        'metadata':{
-            'publication_format':'selected_volume_chapter', 'bibliographic_year':1994,
-            'related_workshop_year':1993, 'book_doi':BOOK, 'book_title':SCOPE['book_title'],
-            'year_basis':'Reviewed workshop year; bibliographic publication year is separately retained.',
-            'first_page':chapter['first_page'], 'last_page':chapter['last_page'],
-            'authors_citation_raw':chapter['authors_citation_raw'],
-            'publication_evidence':'Published selected-volume chapter, not an original workshop decision note.',
-            'scope_relationship':SCOPE['relationship'], 'decision_date_precision':'not_supplied',
-            'source_evidence':evidence, 'scope_evidence':scope_evidence,
-        },
-    }
-
-
-def collect(out, *, toc_snapshots, chapter_snapshots):
-    toc_index, chapter_index = snapshot_index(toc_snapshots), snapshot_index(chapter_snapshots)
-    frozen = {}
-    def read(index, url):
-        raw, evidence = read_page(index, url)
-        frozen[evidence['file']] = raw
-        return raw.decode('utf-8'), evidence
-    history, history_evidence = read(toc_index, HISTORY_URL)
-    if hashlib.sha256(history.encode()).hexdigest() != HISTORY_SHA:
-        raise ValueError('Reviewed selected-volume relationship evidence changed')
-    scope_evidence = [history_evidence]
-    titles, excluded = {}, []
-    for page in range(1, 4):
-        html, evidence = read(toc_index, BOOK_URL+('?page='+str(page) if page > 1 else ''))
-        scope_evidence.append(evidence)
-        entries = parse_toc(html, excluded_materials=excluded)
-        if titles.keys() & entries.keys():
-            raise ValueError('DOI repeated across TOC pages')
-        titles.update(entries)
-    expected = {BOOK+'_'+str(i) for i in range(1, 50)}
-    if titles.keys() != expected:
-        raise ValueError('Reviewed 49-chapter inventory changed')
-    chapter_urls = {'https://link.springer.com/chapter/'+doi for doi in expected}
-    if chapter_index.keys() != chapter_urls:
-        raise ValueError('Missing or extraneous chapter snapshot')
-    records = []
-    for doi in sorted(expected, key=lambda s:int(s.rsplit('_',1)[1])):
-        html, evidence = read(chapter_index, 'https://link.springer.com/chapter/'+doi)
-        chapter = parse_chapter(html, doi)
-        if chapter['title'] != titles[doi]:
-            raise ValueError('Chapter title disagrees with official TOC')
-        records.append(record_for(chapter, evidence, scope_evidence))
-    out = Path(out)
-    out.mkdir(parents=True, exist_ok=True)
-    with (out/'springer-selected.collect.lock').open('a') as lock:
-        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        if any(p.name != 'springer-selected.collect.lock' for p in out.iterdir()):
-            raise FileExistsError('Existing output must be inspected, not overwritten')
-        with tempfile.TemporaryDirectory(prefix='.springer-selected-', dir=out) as temporary:
-            stage = Path(temporary)
-            (stage/'evidence').mkdir()
-            for filename, raw in frozen.items():
-                (stage/filename).write_bytes(raw)
-            path = stage/'springer-aistats-1993-selected-volume.jsonl'
-            path.write_text(''.join(json.dumps(r,ensure_ascii=False)+'\n' for r in records))
-            manifest = {'source':SOURCE, 'parser_version':1,
-                        **{k:SCOPE[k] for k in ('venue','year','track','event_key')},
-                        'count':len(records), 'file':str(out/path.name),
-                        'file_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
-                        'list_url':BOOK_URL, 'status':'partial',
-                        'excluded_materials':excluded,
-                        'missing_abstract_count':sum(not r['abstract'] for r in records)}
-            path.with_suffix('.manifest.json').write_text(json.dumps(manifest,indent=2)+'\n')
-            report = {'status':'collected_pending_identity_review', 'records':len(records),
-                      'files':[manifest], 'reviewed_scope':SCOPE, 'database_written':False,
-                      'pdfs_downloaded':False, 'scope_evidence':scope_evidence}
-            (stage/'collection-report.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
-            for item in stage.iterdir():
-                if item.name != 'collection-report.json':
-                    item.replace(out/item.name)
-            (stage/'collection-report.json').replace(out/'collection-report.json')
-    return report
-
-
-def main(argv=None, *, prog=None):
-    parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--toc-snapshots', required=True)
-    parser.add_argument('--chapter-snapshots', required=True)
-    parser.add_argument('--out', required=True)
-    args = parser.parse_args(argv)
-    print(json.dumps(collect(args.out, toc_snapshots=args.toc_snapshots,
-                             chapter_snapshots=args.chapter_snapshots),ensure_ascii=False))

tests/test_dblp_recovery.py

@@ -0,0 +1,141 @@
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.dblp import catalogue
+from paper_atlas.collectors.dblp_recovery import recover
+from paper_atlas.collectors.dblp_scope import scope
+from paper_atlas.ingestion.records import ingest_file
+
+
+class DblpRecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.dtd = self.root / 'dblp.dtd'
+        self.dtd.write_bytes(b'<!ENTITY eacute "&#233;">')
+        self.archive = self.root / 'dblp.xml.gz'
+        # These include a keynote and a workshop, neither a main-paper inference.
+        body = '''<proceedings key="conf/ifaamas/2025"><title>AAMAS 2025</title><year>2025</year></proceedings>
+<inproceedings key="conf/ifaamas/keynote"><author>Ren&eacute;</author><title>Keynote: Agents</title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref></inproceedings>
+<inproceedings key="conf/ifaamas/paper"><author>Alice Doe</author><title>Research <i>agents</i></title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref><ee>https://dl.acm.org/doi/10.5555/3709347.123</ee></inproceedings>
+<proceedings key="conf/ifaamas/2025w"><title>AAMAS Workshop</title><year>2025</year></proceedings>
+<inproceedings key="conf/ifaamas/workshop"><title>Workshop paper</title><year>2025</year><booktitle>AAMAS Workshop</booktitle><crossref>conf/ifaamas/2025w</crossref></inproceedings>
+<inproceedings key="conf/unrelated/other"><title>Other family</title><year>2025</year></inproceedings>'''
+        with gzip.open(self.archive, 'wb') as stream:
+            stream.write(('<?xml version="1.0"?><!DOCTYPE dblp SYSTEM "dblp.dtd"><dblp>'
+                          + body + '</dblp>').encode())
+        self.sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+        self.md5 = hashlib.md5(self.dtd.read_bytes()).hexdigest()
+        self.manifest = catalogue(self.archive, self.dtd, self.root / 'catalogue',
+                                  self.sha, self.md5)
+        self.catalogue = Path(self.manifest['catalogue'])
+        self.manifest_path = self.catalogue.with_name('dblp-ai-candidates.manifest.json')
+
+    def run_recovery(self, **kwargs):
+        options = dict(expected_sha=self.sha, expected_dtd_md5=self.md5)
+        options.update(kwargs)
+        return recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
+                       ['ifaamas'], **options)
+
+    def test_fresh_discovery_and_recovery_from_old_candidate_omission(self):
+        self.assertEqual(self.manifest['candidate_family_counts']['ifaamas'], 3)
+        # Simulate an older filter without altering its complete catalogue.
+        manifest = dict(self.manifest)
+        manifest['candidate_family_counts'] = {}
+        self.manifest_path.write_text(json.dumps(manifest))
+        candidate = Path(manifest['file'])
+        candidate.write_text('old incomplete candidate file\n')
+        before = {p: p.read_bytes() for p in [self.archive, self.dtd, self.catalogue,
+                                             self.manifest_path, candidate]}
+        with patch('urllib.request.urlopen', side_effect=AssertionError('No network')):
+            report = self.run_recovery()
+        rows = [json.loads(line) for line in Path(report['file']).read_text().splitlines()]
+        self.assertEqual(report['recovered_publications'], 3)
+        self.assertEqual(report['recovered_volumes'], 2)
+        self.assertEqual(report['recovered_records'], 5)
+        self.assertEqual(report['records_scanned'], 6)
+        self.assertEqual(report['old_candidate_family_counts'], {'ifaamas': 0})
+        self.assertFalse(report['database_written'])
+        self.assertEqual(next(r for r in rows if r['key'].endswith('/keynote'))['fields']['author'], ['René'])
+        self.assertTrue(any(r['key'].endswith('/workshop') for r in rows))
+        self.assertTrue(all(not ({'source', 'status', 'venue', 'track'} & r.keys()) for r in rows))
+        publication = next(r for r in rows if r['key'].endswith('/paper'))
+        volume = next(r for r in rows if r['key'] == 'conf/ifaamas/2025')
+        self.assertEqual(scope(publication, volume), (None, 'outside_target_family'))
+        self.assertEqual({p: p.read_bytes() for p in before}, before)
+        self.assertEqual(hashlib.sha256(Path(report['file']).read_bytes()).hexdigest(),
+                         report['file_sha256'])
+        # Raw recovery files cannot accidentally enter the normalized writer.
+        with patch('paper_atlas.ingestion.records.open_store') as writer:
+            with self.assertRaisesRegex(ValueError, 'source manifest'):
+                ingest_file(self.root / 'never.sqlite', report['file'])
+            writer.assert_not_called()
+        self.assertFalse((self.root / 'never.sqlite').exists())
+
+    def test_checksum_mismatch_does_not_publish(self):
+        for options in [{'expected_sha': '0' * 64}, {'expected_dtd_md5': '0' * 32}]:
+            with self.subTest(options=options), self.assertRaisesRegex(ValueError, 'checksum'):
+                self.run_recovery(**options)
+            self.assertFalse((self.root / 'recovered').exists())
+
+    def test_wrong_catalogue_snapshot_and_incomplete_totals_are_rejected(self):
+        for changes in [{'snapshot_sha256': '0' * 64}, {'dtd_md5': '0' * 32},
+                        {'records_scanned': 99}, {'record_types': {'proceedings': 2}}]:
+            manifest = {**self.manifest, **changes}
+            self.manifest_path.write_text(json.dumps(manifest))
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                self.run_recovery()
+            self.assertFalse((self.root / 'recovered').exists())
+
+    def test_changed_family_count_does_not_publish_partial_result(self):
+        with sqlite3.connect(self.catalogue) as c:
+            c.execute("UPDATE family_counts SET count=count+1 WHERE family='ifaamas' AND record_type='inproceedings'")
+        with self.assertRaisesRegex(ValueError, 'family totals'):
+            self.run_recovery()
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_changed_volume_metadata_is_detected(self):
+        with sqlite3.connect(self.catalogue) as c:
+            row = json.loads(c.execute("SELECT metadata_json FROM proceedings WHERE key='conf/ifaamas/2025'").fetchone()[0])
+            row['fields']['title'] = ['Wrong volume']
+            c.execute("UPDATE proceedings SET metadata_json=? WHERE key='conf/ifaamas/2025'", (json.dumps(row),))
+        with self.assertRaisesRegex(ValueError, 'proceedings differs'):
+            self.run_recovery()
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_truncated_gzip_never_publishes(self):
+        self.archive.write_bytes(self.archive.read_bytes()[:-8])
+        revised_sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+        self.manifest_path.write_text(json.dumps({**self.manifest, 'snapshot_sha256': revised_sha}))
+        with self.assertRaises((EOFError, OSError)):
+            self.run_recovery(expected_sha=revised_sha)
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_existing_output_is_preserved_and_unknown_family_is_rejected(self):
+        out = self.root / 'recovered'
+        out.mkdir()
+        sentinel = out / 'keep'
+        sentinel.write_text('existing')
+        with self.assertRaisesRegex(ValueError, 'already exist'):
+            self.run_recovery()
+        self.assertEqual(sentinel.read_text(), 'existing')
+        for names in [[], ['../ifaamas'], ['not-a-supported-family'], [None]]:
+            with self.subTest(names=names), self.assertRaises(ValueError):
+                recover(self.archive, self.dtd, self.catalogue, self.root / 'other', names)
+
+    def test_requested_family_missing_from_catalogue_is_explicit(self):
+        with self.assertRaisesRegex(ValueError, 'absent from complete catalogue'):
+            recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
+                    ['ifaamas', 'icml'], expected_sha=self.sha, expected_dtd_md5=self.md5)
+        self.assertFalse((self.root / 'recovered').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_springer_selected.py

@@ -1,160 +0,0 @@
-"""Real 49-chapter snapshots compared with a separately frozen metadata audit."""
-import copy
-import gzip
-import hashlib
-import json
-from pathlib import Path
-import sqlite3
-import tempfile
-import unittest
-
-from paper_atlas.collectors.springer_selected import (
-    BOOK, SOURCE, collect, parse_chapter, parse_toc, read_page, snapshot_index,
-)
-from paper_atlas.db.store import Store
-from paper_atlas.ingestion.records import ingest_file
-
-FIXTURE = Path(__file__).parent/'fixtures/springer-aistats1993-fixture.json.gz'
-
-
-class SpringerSelectedTests(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        raw = FIXTURE.read_bytes()
-        if hashlib.sha256(raw).hexdigest() != '94028fd37080f9d8027e5790e379cc151ee6eab17ed0d1dcc0f59e6e75f9d4d0':
-            raise AssertionError('Fixed source fixture differs')
-        cls.fixture = json.loads(gzip.decompress(raw))
-
-    def snapshots(self, root):
-        for group in ['toc','chapters']:
-            folder = root/group
-            folder.mkdir()
-            for name, text in self.fixture[group].items():
-                (folder/name).write_text(text)
-        return {'toc_snapshots':root/'toc','chapter_snapshots':root/'chapters'}
-
-    def test_all49_original_fields_against_independent_audit(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            report = collect(root/'out', **self.snapshots(root))
-            manifest, = report['files']
-            rows = [json.loads(s) for s in Path(manifest['file']).read_text().splitlines()]
-            self.assertEqual(len(rows), 49)
-            self.assertEqual(manifest['status'], 'partial')
-            self.assertEqual(manifest['missing_abstract_count'], 0)
-            self.assertTrue(manifest['excluded_materials'])
-            self.assertTrue(all(x['title'] == 'Front Matter' for x in manifest['excluded_materials']))
-            for row, expected in zip(rows, self.fixture['expected']):
-                for field in ['title','abstract']:
-                    self.assertEqual(row[field], expected[field])
-                self.assertEqual(row['identifiers'], {'doi':expected['doi']})
-                self.assertEqual(row['submission_identifiers'], [{'scheme':'publication_doi','value':expected['doi']}])
-                self.assertEqual(row['external_id'], expected['doi'])
-                self.assertEqual(row['metadata']['authors_citation_raw'], expected['authors_citation_raw'])
-                self.assertEqual(row['authors'], [{'family':s.split(',',1)[0].strip(), 'given':s.split(',',1)[1].strip()} for s in expected['authors_citation_raw']])
-                self.assertEqual(row['metadata']['first_page'], expected['first_page'])
-                self.assertEqual(row['metadata']['last_page'], expected['last_page'])
-                self.assertEqual((row['year'],row['metadata']['bibliographic_year']), (1993,1994))
-                self.assertEqual(row['track'], 'selected_volume')
-                self.assertEqual(row['event_key'], 'aistats-1993-selected-volume-1994')
-                self.assertNotIn('decision_date', row)
-                self.assertIn('original 1993 review decision not supplied', row['raw_decision'])
-                for e in [row['metadata']['source_evidence'], *row['metadata']['scope_evidence']]:
-                    self.assertEqual(hashlib.sha256((root/'out'/e['file']).read_bytes()).hexdigest(), e['sha256'])
-            self.assertFalse(report['database_written'])
-            with self.assertRaises(FileExistsError):
-                collect(root/'out', toc_snapshots=root/'toc', chapter_snapshots=root/'chapters')
-
-    def test_native_history_namespaces_and_idempotent_import(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            report = collect(root/'out', **self.snapshots(root))
-            manifest, = report['files']
-            first = json.loads(Path(manifest['file']).read_text().splitlines()[0])
-            db = root/'sample.sqlite'
-            store = Store(db)
-            pid, rid = store.record('prior_public','earlier','https://example.org/earlier',first['title'],first['abstract'],first['authors'])
-            store.decision(pid,rid,'AISTATS',1993,'main','earlier','https://example.org/earlier','rejected','original public rejection')
-            # Explicitly verified work identity; the short title alone must not
-            # force a merge or attach this publication DOI to the old attempt.
-            store.conn.execute('INSERT INTO identifiers VALUES(?,?,?)', ('doi',first['identifiers']['doi'],pid))
-            store.conn.commit()
-            old = store.conn.execute('SELECT * FROM decisions').fetchall()
-            store.conn.close()
-            ingest_file(db, manifest['file'])
-            ingest_file(db, manifest['file'])
-            with sqlite3.connect(db) as c:
-                self.assertEqual(c.execute('SELECT * FROM decisions WHERE status=\'rejected\'').fetchall(), [tuple(r) for r in old])
-                self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 49)
-                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 50)
-                self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0], 50)
-                self.assertEqual(c.execute('SELECT e.year,e.track,e.event_key FROM editions e ORDER BY e.id').fetchall(), [(1993,'main',''),(1993,'selected_volume','aistats-1993-selected-volume-1994')])
-                self.assertEqual(c.execute('SELECT count(*) FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace=?', (SOURCE+':publication_format',)).fetchone()[0], 49)
-                self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers WHERE scheme=\'publication_doi\'').fetchone()[0], 49)
-                self.assertIsNone(c.execute("SELECT 1 FROM submission_identifiers si JOIN submissions s ON s.id=si.submission_id JOIN editions e ON e.id=s.edition_id WHERE e.track='main'").fetchone())
-
-    def test_short_unanchored_title_keeps_identity_unconfirmed(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            store = Store(Path(tmp)/'identity.sqlite')
-            title = 'Statistical strategy: step 1'
-            authors = [{'given':'D. J.','family':'Hand'}]
-            old, _ = store.record('prior_public','earlier','https://example.org/old',title,'Same abstract',authors)
-            new, _ = store.record(SOURCE,BOOK+'_1','https://link.springer.com/chapter/'+BOOK+'_1',title,'Same abstract',authors,{'doi':BOOK+'_1'})
-            self.assertNotEqual(old, new)
-            self.assertEqual(store.conn.execute('SELECT count(*) FROM match_candidates WHERE method=\'exact_title_unconfirmed\'').fetchone()[0], 1)
-            store.conn.close()
-
-    def test_receipt_hash_path_redirect_and_content_type_guards(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            self.snapshots(root)
-            index = snapshot_index(root/'chapters')
-            url = 'https://link.springer.com/chapter/'+BOOK+'_1'
-            self.assertTrue(read_page(index,url)[0])
-            for key, value in [('sha256','0'*64),('file','../escape.html'),('final_url','https://example.org/chapter'),('http_status',403),('headers',{'Content-Type':'application/pdf'})]:
-                changed = copy.deepcopy(index)
-                changed[url][1][key] = value
-                with self.subTest(key=key), self.assertRaises((ValueError,KeyError)):
-                    read_page(changed,url)
-            changed = copy.deepcopy(index)
-            changed[url][1]['final_url'] = url+'?page=2'
-            with self.assertRaises(ValueError):
-                read_page(changed,url)
-
-    def test_missing_or_duplicate_snapshot_stops_collection(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            args = self.snapshots(root)
-            path = root/'chapters/receipts.json'
-            d = json.loads(path.read_bytes())
-            original = copy.deepcopy(d)
-            d['responses'].pop()
-            path.write_text(json.dumps(d))
-            with self.assertRaises(ValueError):
-                collect(root/'out', **args)
-            self.assertFalse((root/'out').exists())
-            original['responses'].append(original['responses'][0])
-            path.write_text(json.dumps(original))
-            with self.assertRaises(ValueError):
-                snapshot_index(root/'chapters')
-
-    def test_book_doi_year_and_duplicate_citation_fields_reject_drift(self):
-        html = self.fixture['chapters']['chapter-01.html']
-        for key, value in [('citation_doi',BOOK+'_2'),('citation_inbook_title','Another Book'),('citation_publication_date','1993')]:
-            import re
-            changed, count = re.subn(r'(<meta[^>]*name="'+key+r'"[^>]*content=")[^"]*', lambda m:m[1]+value, html)
-            self.assertEqual(count, 1)
-            with self.subTest(key=key), self.assertRaises(ValueError):
-                parse_chapter(changed, BOOK+'_1')
-        with self.assertRaises(ValueError):
-            parse_chapter(html+'<meta name="citation_doi" content="'+BOOK+'_1">', BOOK+'_1')
-
-    def test_missing_abstract_is_explicit_and_does_not_use_other_page_text(self):
-        html = self.fixture['chapters']['chapter-01.html'].replace('id="Abs1-content"','id="not-an-abstract"')
-        self.assertIsNone(parse_chapter(html, BOOK+'_1')['abstract'])
-
-    def test_toc_ignores_frontmatter_but_rejects_duplicate_publications(self):
-        html = self.fixture['toc']['springer-toc1.html']
-        self.assertEqual(len(parse_toc(html)), 17)
-        with self.assertRaises(ValueError):
-            parse_toc(html+html)