Audit raw arXiv IDs and preserve per-record category evidence

Reviewed-on: #2

docs/data-sources.md

@@ -232,8 +232,20 @@ python3 -B -m paper_atlas collect ecva \
   --cache /data/paper-index/raw/http-cache --out /data/paper-index/collected
 ```
 
-CVF 从官方根目录发现 CVPR、ICCV 主会与 Findings，分别保存 track；不把 workshop 或 WACV 算入这些会议。优先跟随页面明确提供的 All Papers 链接，避免只收录默认显示的第一天。ECVA 从官方总目录按论文 URL 的会议与年份分组，当前总目录列有 ECCV 2018、2020、2022、2024。不存在于来源的 `--venue`、`--year` 或二者组合会报错，不补造双年会议届次，也不能因此推断旧届次未举办。
+CVF 从官方根目录发现 CVPR、ICCV 主会与 Findings，分别保存 track；不把 workshop 或 WACV 算入这些会议。优先跟随页面明确提供的 All Papers 链接；只有逐日目录时遍历全部明确链接的日期，再按论文 URL 合并重复条目并检查标题、作者、DOI 一致性。`.py` 与无后缀的同届入口可关联，不跟随其他会议、年份或站点的日期链接。ECVA 从官方总目录按论文 URL 的会议与年份分组，当前总目录列有 ECCV 2018、2020、2022、2024。不存在于来源的 `--venue`、`--year` 或二者组合会报错，不补造双年会议届次，也不能因此推断旧届次未举办。
 
 只访问论文索引与摘要 HTML/PHP 页面。作者优先取 citation 元数据，旧页面尊重声明编码；无 citation 作者时取索引中的作者行，排除论文集出版信息。DOI 只取该论文条目所属的 DOI 链接或 citation 字段；冲突会失败，不从摘要引用中抽取当前论文 ID。不请求 PDF、补充材料、视频或代码。
 
 每份清单记录索引入口数、论文数、缺失摘要与作者数，缺失内容保留为空并标记 partial。存在未被标题条目解析的论文链接时会失败，防止静默漏项。逐来源的 discovery、index-status、last-run 文件分别保存发现目录、逐届结果和实际请求范围。可重复使用 `--year` 或 `--venue` 限定重试；其他届次状态会保留。官方开放目录之外的早期届次、尚未刊出的新届次和独立出版方历史数据仍需后续补齐，采集器完成不代表全历史覆盖已完成。
+
+单个摘要页面 404 时，仍保存官方目录提供的题目、作者和发表证据，摘要留空并记录失败 URL；整届文件标为 partial。403 等访问拒绝会终止该届采集。刷新缺失摘要时，存储层保留原来已知的论文摘要及其来源观察信息，后续页面恢复时可以更新。
+
+针对旧 catalogue 的 blocked 届次可单独补采到新目录，避免与现有采集任务同时改写状态文件：
+
+```sh
+paper-atlas collect cvf --cache /data/paper-index/raw/http-cache \
+  --retry-catalogue /data/paper-index/collected/cvf-index-status.json \
+  --out /data/paper-index/collected/cvf-repair
+```
+
+parser_version 2 的清单还保存实际遍历的 index_urls、入口错误及缺失摘要页面。补采结果经过清单与身份核验后才导入，不能因重试任务启动就标记缺口已解决。

docs/operations.md

@@ -2,6 +2,24 @@
 
 代码以固定 commit 部署，已经启动的任务继续使用其原版本。目录重构不要求迁移数据库，也不更改已有 JSONL 和 manifest 格式。
 
+## arXiv 原始编号与类别证据
+
+原始行数、唯一 arXiv 编号数和合并后的论文实体数分别核验。以下只读原始压缩包，流式处理到 ZIP EOF/CRC 检查完成；输出审计 JSON 与编号 SQLite 索引，不修改论文主库：
+
+```sh
+paper-atlas audit arxiv --archive /data/paper-index/raw/arxiv-metadata.zip \
+  --out /data/paper-index/collected/arxiv-source-audit.json
+```
+
+同编号的重复行会记录出现位置、次数和内容哈希版本，不直接视作不同论文；同篇论文的多个 arXiv 编号则保留各自来源及类别证据。旧导入器使用共享类别证据键，可能在论文实体合并时丢失某个编号的相同类别出处。新导入使用逐记录证据键；已有数据须在没有其他主库写入任务时按保留的来源 metadata 修复：
+
+```sh
+paper-atlas maintain arxiv-tags --db /data/paper-index/papers.sqlite \
+  --out /data/paper-index/collected/arxiv-tag-repair.json
+```
+
+修复使用同一维护锁，分批提交并保存断点，保留 `custom:*` 和其他来源证据；失败后确认原因已解决，再增加 `--resume` 从上次提交位置继续。完成报告不能作为重新扫描的断点，重新核验应指定新报告。源审计和标签修复均保留 8 GiB 空闲空间，不能代替所有来源入库后的实体、投稿去重与完整性检查。
+
 ## 长期维护与更新
 
 1. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。
@@ -16,6 +34,16 @@
 
 2026-11-06 从目标服务器测试：Kaggle 元数据 GET 可下载；OpenReview 会议 group 接口可读，但 notes 接口返回 `ChallengeRequiredError`。这限制了直接补全；已使用来源可追溯的公开历史快照补充近期投稿与明确决定。仍不能用录用名单替代全部投稿，也不能声称覆盖了全部拒稿。
 
-已核验的采集文件计数：ICLR 2020–2026 主会 15,534 条、COLM 2024–2026 共 1,573 条，摘要均已取得；候选范围内的 PMLR 文件为 31,428 条，已剔除两份重复出版元数据。ACL Anthology 已选文件为 63,297 条，其中 15,147 条源记录缺摘要。这些是采集文件的来源记录数，不能相加当作去重论文数，也不能代替主库实时统计；实际入库量使用 paper-atlas query --stats 查看。
+## 查询批次与入库状态
 
-代码在 `codex/metadata-index` 工作分支交付，[Draft PR #1](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls/1) 尚未合并，等待全量就绪后用户手动合并。各提交分别由远程 CI 验证，结果以对应提交的 Gitea Actions 检查为准。全量导入、其他会议适配器、去重复核和覆盖率核验继续进行，应随实际结果更新此说明。
+采集文件、已入库的来源记录和去重论文实体是不同的计数。来源可能覆盖同一篇论文，同一篇论文也可能有多次投稿；不能把各批文件数量相加作为独立论文总数。实时统计与覆盖报告使用：
+
+```sh
+paper-atlas query --db /data/paper-index/papers.sqlite --stats
+paper-atlas coverage --db /data/paper-index/papers.sqlite \
+  --out /data/paper-index/collected/venue-coverage-current.json
+```
+
+manifest 和采集器状态清单用于确认文件是否齐全及是否缺摘要；`ingest_runs` 和协调器日志用于确认主库导入是否成功。官方程序、书目目录和公开投稿快照分别保留，未出现在某份清单不等于被拒稿；第三方快照也不能代表平台当前全量状态。代码测试成功不能代替最终全库去重、完整性和逐届覆盖核验。
+
+当前代码与合并状态以 [Pull Requests](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls) 为准；CI 结果以对应提交的 Actions 检查为准。工作分支交付后等待用户手动合并。合并代码不自动部署、不自动启动采集，也不代表数据批次已完成。

docs/usage.md

@@ -42,4 +42,13 @@ paper-atlas coverage --db /data/paper-index/papers.sqlite --out /data/paper-inde
 
 `--status rejected` 筛选某次投稿的最新已知决定。它并不表示论文在所有会议均未录用；分析转投情况时查看该论文的完整投稿历史。年份表示会议届次，观察时间和决定日期分别保留。
 
+例如，查找在 ICLR 有拒稿记录、但在其他已知投稿中已有录用的论文：
+
+```sh
+paper-atlas query --db /data/paper-index/papers.sqlite \\
+  --venue ICLR --status rejected --outcome accepted
+```
+
+`known_acceptances` 列出目前仍为录用状态的各次投稿及会议、年份、track 和决定证据，`publication_outcome` 汇总为 accepted 或 no_known_acceptance。汇总会检查所有会议和 track，不把某次拒稿覆盖到整篇论文；同一次投稿的后续撤稿也会更新该次状态并保留原决定。no_known_acceptance 只表示现有来源中没有查到录用，不能称为“从未录用”。
+
 `--stats` 中 papers 是当前规则下的论文实体数；source_records 是来源记录数；decisions 是决定观察数。一个投稿可能有多个来源和多条观察，不能相加当作论文数。

paper_atlas/audits/arxiv.py

@@ -0,0 +1,90 @@
+"""Independently count raw arXiv IDs and their payload variants to ZIP EOF."""
+import argparse
+from contextlib import closing
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import sqlite3
+import tempfile
+import zipfile
+
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.snapshots import digest_file
+
+MEMBER='arxiv-metadata-oai-snapshot.json'
+
+
+def fingerprint(path):
+    info=path.stat()
+    return {'bytes':info.st_size,'mtime_ns':info.st_mtime_ns}
+
+
+def audit(archive,out,expected_sha256=None):
+    out=Path(out).resolve();out.parent.mkdir(parents=True,exist_ok=True)
+    with out.with_suffix('.lock').open('a') as lock:
+        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
+        return _audit(archive,out,expected_sha256)
+
+
+def _audit(archive,out,expected_sha256=None):
+    archive=Path(archive).resolve();out=Path(out).resolve()
+    ids=out.with_suffix('.ids.sqlite')
+    if out.exists() or ids.exists():raise ValueError('Audit output exists; inspect it before another run')
+    out.parent.mkdir(parents=True,exist_ok=True)
+    before=fingerprint(archive);identity=digest_file(archive)
+    if expected_sha256 and identity['sha256']!=expected_sha256:raise ValueError('Archive checksum mismatch')
+    if shutil.disk_usage(out.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free before source audit')
+    count=0;missing_title=0;missing_abstract=0
+    with tempfile.TemporaryDirectory(prefix='.arxiv-audit-',dir=out.parent) as temporary:
+        stage=Path(temporary)/ids.name
+        with closing(sqlite3.connect(stage)) as c:
+            c.executescript('PRAGMA cache_size=-16384; CREATE TABLE raw_ids(external_id TEXT PRIMARY KEY,first_line INTEGER,last_line INTEGER,occurrences INTEGER,first_sha256 BLOB,has_payload_variants INTEGER) WITHOUT ROWID; CREATE TABLE duplicate_variants(external_id TEXT,payload_sha256 BLOB,first_line INTEGER,last_line INTEGER,occurrences INTEGER,PRIMARY KEY(external_id,payload_sha256)) WITHOUT ROWID;')
+            with zipfile.ZipFile(archive) as zipped:
+                info=zipped.getinfo(MEMBER)
+                with zipped.open(info) as stream:
+                    for count,line in enumerate(stream,1):
+                        row=json.loads(line);external=row.get('id')
+                        if not isinstance(external,str) or not external or external!=external.strip():raise ValueError('Missing or malformed arXiv ID')
+                        payload=hashlib.sha256(json.dumps(row,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()).digest()
+                        inserted=c.execute('INSERT OR IGNORE INTO raw_ids VALUES(?,?,?,1,?,0)',(external,count,count,payload)).rowcount
+                        if not inserted:
+                            first_line,first=c.execute('SELECT first_line,first_sha256 FROM raw_ids WHERE external_id=?',(external,)).fetchone()
+                            c.execute('INSERT OR IGNORE INTO duplicate_variants VALUES(?,?,?,?,1)',(external,first,first_line,first_line))
+                            c.execute('INSERT INTO duplicate_variants VALUES(?,?,?,?,1) ON CONFLICT(external_id,payload_sha256) DO UPDATE SET last_line=excluded.last_line,occurrences=duplicate_variants.occurrences+1',(external,payload,count,count))
+                            c.execute('UPDATE raw_ids SET last_line=?,occurrences=occurrences+1,has_payload_variants=has_payload_variants OR first_sha256!=? WHERE external_id=?',(count,payload,external))
+                        missing_title+=not bool((row.get('title') or '').strip())
+                        missing_abstract+=not bool((row.get('abstract') or '').strip())
+                        if count%10000==0:
+                            c.commit()
+                            if shutil.disk_usage(out.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free during source audit')
+                        if count%100000==0:print(json.dumps({'source':'arxiv_audit','raw_rows':count}),flush=True)
+                # zipfile verifies the member CRC only when its stream reaches EOF.
+            if not count:raise ValueError('Empty arXiv metadata member')
+            if fingerprint(archive)!=before:raise ValueError('Archive changed during audit')
+            c.commit()
+            if c.execute('PRAGMA quick_check').fetchone()[0]!='ok':raise ValueError('Audit index failed SQLite quick_check')
+            unique=c.execute('SELECT count(*) FROM raw_ids').fetchone()[0]
+            duplicate_ids=[]
+            for key,first,last,occurrences,variants in c.execute('SELECT external_id,first_line,last_line,occurrences,has_payload_variants FROM raw_ids WHERE occurrences>1 ORDER BY external_id'):
+                duplicate_ids.append({'external_id':key,'first_line':first,'last_line':last,'occurrences':occurrences,'has_payload_variants':bool(variants),
+                                      'variants':[{'sha256':digest.hex(),'first_line':a,'last_line':b,'occurrences':n} for digest,a,b,n in c.execute('SELECT payload_sha256,first_line,last_line,occurrences FROM duplicate_variants WHERE external_id=? ORDER BY first_line',(key,))]})
+            result={'checked_at':timestamp(),'archive':str(archive),'archive_identity':identity,'member':MEMBER,'member_crc32':f'{info.CRC:08x}',
+                    'uncompressed_bytes':info.file_size,'zip_eof_crc_verified':True,'raw_rows':count,'unique_external_ids':unique,'extra_rows_for_repeated_ids':count-unique,
+                    'repeated_id_count':len(duplicate_ids),'repeated_ids_with_payload_variants':sum(item['has_payload_variants'] for item in duplicate_ids),
+                    'missing_title_rows':missing_title,'missing_abstract_rows':missing_abstract,'duplicate_ids':duplicate_ids,'id_index':str(ids),
+                    'interpretation':'Raw ID/payload audit, not a count of unique research works. Payload variants are retained as evidence; latest source import uses the last row for each ID. Cross-source entity/submission dedup requires a separate database audit.'}
+        stage.replace(ids)
+        part=out.with_suffix('.part');part.write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');part.replace(out)
+    print(json.dumps({key:value for key,value in result.items() if key!='duplicate_ids'}),flush=True)
+    return result
+
+
+def main(argv=None,*,prog=None):
+    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--archive',required=True);parser.add_argument('--out',required=True)
+    parser.add_argument('--expected-sha256');args=parser.parse_args(argv)
+    audit(args.archive,args.out,args.expected_sha256)
+
+
+if __name__=='__main__':main()

paper_atlas/cli.py

@@ -24,7 +24,9 @@ COMMANDS = {
     ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),
     ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
     ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
+    ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
     ('audit','iclr'): ('paper_atlas.audits.iclr', {}, '核对 ICLR 快照与同届官方程序'),
+    ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'),
 }
 
 

paper_atlas/collectors/vision.py

@@ -5,6 +5,8 @@ import json
 from pathlib import Path
 import re
 import time
+from datetime import date
+from urllib.error import HTTPError
 from urllib.parse import parse_qs, unquote, urljoin, urlsplit
 
 from paper_atlas.io.http import fetch
@@ -98,6 +100,14 @@ def discover(source,cache):
     return sorted(indexes.values(),key=lambda row:(row['year'],row['venue'],row['track']!='main',row['track']))
 
 
+def author_names(raw):
+    raw=(raw or '').split(';',1)[0]
+    if ' and ' in raw and all(',' in name for name in raw.split(' and ')):
+        names=[' '.join(reversed([part.strip() for part in name.split(',',1)])) for name in raw.split(' and ')]
+    else:names=raw.split(',')
+    return [clean(name).removesuffix('*').strip() for name in names if clean(name).removesuffix('*').strip()]
+
+
 def parse_paper(html,entry,source,venue,year,track,index_url):
     nodes=list(Document(html).root.walk());meta=defaultdict(list)
     for node in nodes:
@@ -112,11 +122,7 @@ def parse_paper(html,entry,source,venue,year,track,index_url):
         if clean(author):authors.append(clean(author))
     if not authors:
         raw=next((clean(n.text()) for n in nodes if n.attrs.get('id')=='authors'),'') or entry.get('index_authors') or ''
-        raw=raw.split(';',1)[0]
-        if ' and ' in raw and all(',' in name for name in raw.split(' and ')):
-            names=[' '.join(reversed([part.strip() for part in name.split(',',1)])) for name in raw.split(' and ')]
-        else:names=raw.split(',')
-        authors=[clean(name).removesuffix('*').strip() for name in names if clean(name).removesuffix('*').strip()]
+        authors=author_names(raw)
     dois={value for doi in meta['citation_doi'] if doi and (value:=doi_link(doi) or doi.strip().lower()) and re.fullmatch(r'10\.\d{4,9}/\S+',value)}
     if entry.get('doi'):dois.add(entry['doi'])
     if len(dois)>1:raise ValueError('Citation DOI differs from the official index')
@@ -126,35 +132,95 @@ def parse_paper(html,entry,source,venue,year,track,index_url):
             'title':clean(title),'abstract':abstract,'authors':authors,'identifiers':identifiers,
             'venue':venue,'year':year,'track':track,'status':'accepted',
             'raw_decision':'listed in official proceedings','list_url':index_url,
-            'metadata':{'citation':dict(meta),'index_title':entry['index_title'],'index_authors':entry.get('index_authors')}}
+            'metadata':{'citation':dict(meta),'index_title':entry['index_title'],'index_authors':entry.get('index_authors'),
+                        'index_urls':entry.get('index_urls',[index_url])}}
+
+
+def index_identity(url):
+    parsed=urlsplit(url)
+    match=re.fullmatch(r'/(CVPR|ICCV)(20\d{2})(_findings)?(?:\.py)?/?',parsed.path)
+    return match.groups() if parsed.netloc=='openaccess.thecvf.com' and match else None
+
+
+def read_index(index,cache):
+    url=index['index_url'];html=decode_html(fetch(url,cache,32*1024**2,delay=1))
+    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',[])
+        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']
+            except ValueError:valid=False
+            if not valid:continue
+        links.add(link)
+    all_pages=sorted(link for link in links if parse_qs(urlsplit(link).query)['day']==['all'])
+    day_pages=sorted(links-set(all_pages));index_errors=[]
+    if len(all_pages)>1:raise ValueError('Ambiguous all-papers index')
+    pages=[]
+    if all_pages:
+        try:pages=[(all_pages[0],decode_html(fetch(all_pages[0],cache,32*1024**2,delay=1)))]
+        except HTTPError as error:
+            if error.code!=404 or not day_pages:
+                error.close();raise
+            index_errors.append({'url':all_pages[0],'status':404});error.close()
+    if not pages and day_pages:
+        pages=[(page,decode_html(fetch(page,cache,32*1024**2,delay=1))) for page in day_pages]
+    if not pages:pages=[(url,html)]
+    entries={};raw_count=0
+    for page,document in pages:
+        parsed,count=index_entries(document,page);raw_count+=count
+        for entry in parsed:
+            entry['index_urls']=[page];old=entries.get(entry['url'])
+            if old:
+                for key in ('index_title','index_authors','doi'):
+                    if old.get(key) and entry.get(key) and old[key]!=entry[key]:raise ValueError('Conflicting metadata across official day indexes')
+                    if not old.get(key):old[key]=entry.get(key)
+                old['index_urls'].append(page)
+            else:entries[entry['url']]=entry
+    return list(entries.values()),raw_count,[page for page,_ in pages],index_errors
+
+
+def index_only_record(entry,source,venue,year,track,index_url,error):
+    if not entry['index_title'].strip():raise ValueError('Missing title in official index')
+    return {'source':source,'external_id':urlsplit(entry['url']).path,'url':entry['url'],
+            'title':entry['index_title'],'abstract':None,'authors':author_names(entry.get('index_authors')),
+            'identifiers':{'doi':entry['doi']} if entry.get('doi') else {},'venue':venue,'year':year,'track':track,
+            'status':'accepted','raw_decision':'listed in official proceedings index',
+            'evidence_url':entry['index_urls'][0],'list_url':index_url,
+            'metadata':{'index_title':entry['index_title'],'index_authors':entry.get('index_authors'),
+                        'index_urls':entry['index_urls'],'abstract_fetch_error':{'status':error.code,'url':error.url,'message':str(error)}}}
 
 
 def collect_index(source,index,cache,out):
     url=index['index_url'];venue=index['venue'];year=index['year'];track=index['track']
     entries=index.get('entries')
     raw_count=len(entries) if entries is not None else None
+    index_urls=[url];index_errors=[];missing_pages=[]
     if entries is None:
-        html=decode_html(fetch(url,cache,32*1024**2,delay=1))
-        # Some defaults contain only day one; prefer the explicitly linked all-papers page.
-        all_pages={urljoin(url,n.attrs.get('href','')) for n in Document(html).root.walk() if n.tag=='a' and parse_qs(urlsplit(urljoin(url,n.attrs.get('href',''))).query).get('day')==['all'] and urlsplit(urljoin(url,n.attrs.get('href',''))).netloc==urlsplit(url).netloc and urlsplit(urljoin(url,n.attrs.get('href',''))).path==urlsplit(url).path}
-        if len(all_pages)>1:raise ValueError('Ambiguous all-papers index')
-        if all_pages:
-            url=all_pages.pop();html=decode_html(fetch(url,cache,32*1024**2,delay=1))
-        entries,raw_count=index_entries(html,url)
+        entries,raw_count,index_urls,index_errors=read_index(index,cache)
+        if len(index_urls)==1:url=index_urls[0]
     records=[]
     for number,entry in enumerate(entries,1):
         if publication(entry['url'])!=(venue,year):raise ValueError('Publication path differs from requested edition')
-        record=parse_paper(decode_html(fetch(entry['url'],cache,2*1024**2,delay=1)),entry,source,venue,year,track,url)
+        try:record=parse_paper(decode_html(fetch(entry['url'],cache,2*1024**2,delay=1)),entry,source,venue,year,track,url)
+        except HTTPError as error:
+            if error.code!=404:
+                error.close();raise
+            entry.setdefault('index_urls',[url])
+            record=index_only_record(entry,source,venue,year,track,url,error)
+            missing_pages.append(entry['url']);error.close()
         records.append(record)
         if number%25==0:print(json.dumps({'source':source,'venue':venue,'year':year,'track':track,'collected':number,'expected':len(entries)}),flush=True)
     if not records:raise ValueError('Empty publication list')
     path=Path(out)/f'{source}-{venue}{year}-{track}.jsonl';path.parent.mkdir(parents=True,exist_ok=True)
     part=path.with_suffix('.part');part.write_text(''.join(json.dumps(r,ensure_ascii=False)+'\n' for r in records));part.replace(path)
     missing=sum(not r['abstract'] for r in records);missing_authors=sum(not r['authors'] for r in records)
-    manifest={'source':source,'parser_version':1,'venue':venue,'year':year,'track':track,
+    manifest={'source':source,'parser_version':2,'venue':venue,'year':year,'track':track,
               'count':len(records),'index_link_count':len(entries),'raw_index_entry_count':raw_count,
               'missing_abstract_count':missing,'missing_authors_count':missing_authors,'file':str(path),
-              'list_url':url,'status':'partial' if missing or missing_authors else 'collected'}
+              'index_urls':index_urls,'index_errors':index_errors,'missing_abstract_pages':missing_pages,
+              'list_url':url,'status':'partial' if missing or missing_authors or index_errors else 'collected'}
     path.with_suffix('.manifest.json').write_text(json.dumps(manifest,indent=2));print(json.dumps(manifest),flush=True)
     return manifest
 
@@ -165,8 +231,13 @@ def main(argv=None, *, prog=None, source=None):
     else:parser.set_defaults(source=source)
     parser.add_argument('--cache',required=True);parser.add_argument('--out',required=True)
     parser.add_argument('--year',type=int,action='append');parser.add_argument('--venue',choices=['CVPR','ICCV','ECCV'],action='append')
+    parser.add_argument('--retry-catalogue',help='Select only blocked editions from an existing catalogue; write results to --out')
     args=parser.parse_args(argv);out=Path(args.out);out.mkdir(parents=True,exist_ok=True)
     indexes=discover(args.source,args.cache)
+    if args.retry_catalogue:
+        retry=json.loads(Path(args.retry_catalogue).read_text())
+        indexes=[index for index in indexes if retry.get(f'{index["venue"]}:{index["year"]}:{index["track"]}',{}).get('status')=='blocked']
+        if not indexes:parser.error('Retry catalogue has no blocked editions for this source')
     if args.year and set(args.year)-{r['year'] for r in indexes}:parser.error('Requested year is absent from the official source')
     if args.venue and set(args.venue)-{r['venue'] for r in indexes}:parser.error('Requested venue is absent from this source')
     if not any((not args.year or r['year'] in args.year) and (not args.venue or r['venue'] in args.venue) for r in indexes):parser.error('Requested venue/year combination is absent from the official source')
@@ -181,7 +252,7 @@ def main(argv=None, *, prog=None, source=None):
         try:
             manifest=collect_index(args.source,index,args.cache,args.out);catalogue[key]={'status':manifest['status'],'files':[manifest]}
         except Exception as exc:
-            catalogue[key]={'status':'blocked','year':index['year'],'venue':index['venue'],'track':index['track'],'index_url':index['index_url'],'error':str(exc)}
+            catalogue[key]={'status':'blocked','year':index['year'],'venue':index['venue'],'track':index['track'],'index_url':index['index_url'],'error':str(exc),'error_url':getattr(exc,'url',None)}
             print(json.dumps(catalogue[key]),flush=True)
         part=path.with_suffix('.part');part.write_text(json.dumps(catalogue,indent=2));part.replace(path)
     (out/f'{args.source}-last-run.json').write_text(json.dumps({'requested_indexes':requested,'year_filter':args.year,'venue_filter':args.venue},indent=2))

paper_atlas/db/store.py

@@ -88,7 +88,7 @@ class Store:
             found = c.execute('SELECT paper_id FROM identifiers WHERE scheme=? AND value=?',(scheme,value)).fetchone()
             if found:
                 matches.add(found[0])
-        old = c.execute('SELECT paper_id,title,abstract FROM source_records WHERE source_id=? AND external_id=?',(source_id:=self.source(source,url),external_id)).fetchone()
+        old = c.execute('SELECT paper_id,title,abstract,retrieved_at,metadata_json FROM source_records WHERE source_id=? AND external_id=?',(source_id:=self.source(source,url),external_id)).fetchone()
         if old:
             matches.add(old[0])
         key = normalized(title)
@@ -105,13 +105,17 @@ class Store:
             for other in sorted(matches-{paper}):
                 self.merge(paper,other,'explicit identifier or exact title corroborated by authors')
             if old:
-                c.execute("UPDATE papers SET title=?,abstract=?,title_key=?,authors_json=? WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),paper,old[1],old[2]))
+                old_abstract=old[2] or json.loads(old[4]).get('last_known_abstract',{}).get('abstract')
+                c.execute("UPDATE papers SET title=?,abstract=COALESCE(NULLIF(?,''),abstract),title_key=?,authors_json=CASE WHEN ?='[]' THEN authors_json ELSE ? END WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),json.dumps(authors,ensure_ascii=False),paper,old[1],old_abstract))
             c.execute("UPDATE papers SET abstract=CASE WHEN abstract IS NULL OR abstract='' THEN ? ELSE abstract END WHERE id=?",(abstract,paper))
         else:
             paper = c.execute('INSERT INTO papers(title,abstract,title_key,authors_json,first_published) VALUES(?,?,?,?,?)',(title,abstract,key,json.dumps(authors,ensure_ascii=False),published)).lastrowid
         for scheme,value in ids:
             c.execute('INSERT OR IGNORE INTO identifiers VALUES(?,?,?)',(scheme,value,paper))
         metadata=dict(metadata or {});metadata['record_identifiers']=dict(ids)
+        if old and not abstract:
+            previous={'abstract':old[2],'observed_at':old[3]} if old[2] else json.loads(old[4]).get('last_known_abstract')
+            if previous:metadata['last_known_abstract']=previous
         c.execute('INSERT INTO source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(source_id,external_id) DO UPDATE SET paper_id=excluded.paper_id,url=excluded.url,title=excluded.title,abstract=excluded.abstract,authors_json=excluded.authors_json,metadata_json=excluded.metadata_json,retrieved_at=CURRENT_TIMESTAMP',(source_id,external_id,paper,url,title,abstract,json.dumps(authors,ensure_ascii=False),json.dumps(metadata,ensure_ascii=False,default=str)))
         record = c.execute('SELECT id FROM source_records WHERE source_id=? AND external_id=?',(source_id,external_id)).fetchone()[0]
         for other,_,_ in title_matches:

paper_atlas/ingestion/arxiv.py

@@ -10,6 +10,13 @@ from pathlib import Path
 from paper_atlas.db.store import Store
 
 
+def refresh_categories(store,paper,record,categories):
+    """Keep category evidence for each arXiv record, including merged papers."""
+    if not isinstance(categories,str):raise ValueError('arXiv categories must be text')
+    store.record_tags(paper,record,'arxiv',[{'namespace':'arxiv:category','value':value} for value in categories.split()])
+    store.conn.execute("DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key='arxiv:categories' AND tag_id IN (SELECT id FROM tags WHERE namespace='arxiv:category')",(record,))
+
+
 def ingest(db, archive):
     store = Store(db)
     conn = store.conn
@@ -45,11 +52,7 @@ def ingest(db, archive):
                     authors = row.get('authors_parsed') or row.get('authors') or []
                     metadata = {k:v for k,v in row.items() if k not in ('title','abstract','authors','authors_parsed')}
                     paper,record = store.record('arxiv',external_id,'https://arxiv.org/abs/'+external_id,title,abstract,authors,{'arxiv':external_id,'doi':row.get('doi')},metadata,(row.get('versions') or [{}])[0].get('created'))
-                    conn.execute("DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key='arxiv:categories'", (record,))
-                    for category in (row.get('categories') or '').split():
-                        conn.execute("INSERT OR IGNORE INTO tags(namespace,value) VALUES('arxiv:category',?)",(category,))
-                        tag = conn.execute("SELECT id FROM tags WHERE namespace='arxiv:category' AND value=?",(category,)).fetchone()[0]
-                        conn.execute('INSERT OR IGNORE INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,'arxiv:categories'))
+                    refresh_categories(store,paper,record,row.get('categories') or '')
                     if line_number % 500 == 0:
                         if shutil.disk_usage(Path(db).parent).free < 4*1024**3:
                             raise RuntimeError('Stopped before exhausting disk: less than 4 GiB free')

paper_atlas/ingestion/arxiv_tags.py

@@ -0,0 +1,74 @@
+"""Rebuild per-record arXiv category evidence from retained source metadata."""
+import argparse
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import shutil
+import sqlite3
+
+from paper_atlas.db.store import Store
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.ingestion.arxiv import refresh_categories
+
+
+def save(path,state):
+    part=path.with_suffix('.part');part.write_text(json.dumps(state,indent=2)+'\n');part.replace(path)
+
+
+def repair(db,out,resume=False,batch_size=1000):
+    db=Path(db).resolve();out=Path(out).resolve()
+    if not db.is_file():raise ValueError('Existing database required')
+    if batch_size<1 or batch_size>10000:raise ValueError('Batch size must be between 1 and 10000')
+    out.parent.mkdir(parents=True,exist_ok=True)
+    with Path(str(db)+'.maintenance.lock').open('a') as lock:
+        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
+        with closing(sqlite3.connect(db.as_uri()+'?mode=ro',uri=True)) as c:
+            source=c.execute("SELECT id FROM sources WHERE name='arxiv'").fetchone()
+            if not source:raise ValueError('No imported arXiv source')
+            total,maximum=c.execute('SELECT count(*),COALESCE(max(id),0) FROM source_records WHERE source_id=?',source).fetchone()
+        scope={'db':str(db),'source_id':source[0],'source_record_count':total,'maximum_source_record_id':maximum}
+        if resume:
+            state=json.loads(out.read_text())
+            if any(state.get(key)!=value for key,value in scope.items()):raise ValueError('Repair scope changed since checkpoint')
+            if state['status']=='complete':raise ValueError('Completed report exists; use a new report for a fresh repair')
+        else:
+            if out.exists():raise ValueError('Repair report exists; inspect it or explicitly resume')
+            state={**scope,'started_at':timestamp(),'last_record_id':0,'records_refreshed':0}
+        if shutil.disk_usage(db.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free before repair')
+        state.update(status='running',error=None);save(out,state)
+        c=None
+        try:
+            store=Store(db);c=store.conn
+            while True:
+                rows=c.execute('SELECT id,paper_id,metadata_json FROM source_records WHERE source_id=? AND id>? AND id<=? ORDER BY id LIMIT ?',(source[0],state['last_record_id'],maximum,batch_size)).fetchall()
+                if not rows:break
+                for record,paper,payload in rows:
+                    categories=json.loads(payload).get('categories','')
+                    if categories is None:categories=''
+                    refresh_categories(store,paper,record,categories)
+                c.commit()
+                state.update(last_record_id=rows[-1][0],records_refreshed=state['records_refreshed']+len(rows));save(out,state)
+                if state['records_refreshed']%10000<len(rows):print(json.dumps(state),flush=True)
+                if shutil.disk_usage(db.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free during repair')
+            if state['records_refreshed']!=total:raise ValueError('Repaired count differs from fixed source inventory')
+            current=c.execute('SELECT count(*),COALESCE(max(id),0) FROM source_records WHERE source_id=?',source).fetchone()
+            if current!=(total,maximum):raise ValueError('arXiv source inventory changed during repair')
+            legacy=c.execute("SELECT count(*) FROM paper_tags pt JOIN source_records r ON r.id=pt.source_record_id JOIN tags t ON t.id=pt.tag_id WHERE r.source_id=? AND t.namespace='arxiv:category' AND pt.evidence_key='arxiv:categories'",source).fetchone()[0]
+            if legacy:raise ValueError('Legacy category evidence remains after repair')
+            state.update(status='complete',finished_at=timestamp(),legacy_category_evidence_remaining=legacy);save(out,state)
+        except BaseException as error:
+            if c is not None:c.rollback()
+            state.update(status='failed',error=str(error));save(out,state);raise
+        finally:
+            if c is not None:c.close()
+    print(json.dumps(state),flush=True);return state
+
+
+def main(argv=None,*,prog=None):
+    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--db',required=True);parser.add_argument('--out',required=True)
+    parser.add_argument('--resume',action='store_true');parser.add_argument('--batch-size',type=int,default=1000);args=parser.parse_args(argv)
+    repair(args.db,args.out,args.resume,args.batch_size)
+
+
+if __name__=='__main__':main()

paper_atlas/querying.py

@@ -5,6 +5,22 @@ import sqlite3
 from pathlib import Path
 
 
+LATEST_DECISION = '''NOT EXISTS (
+    SELECT 1 FROM decisions newer WHERE newer.submission_id=d.submission_id
+    AND (newer.observed_at,COALESCE(newer.decision_date,newer.observed_at),newer.id)
+      > (d.observed_at,COALESCE(d.decision_date,d.observed_at),d.id)
+)'''
+ACCEPTED_PAPERS = "SELECT s.paper_id FROM submissions s JOIN decisions d ON d.submission_id=s.id WHERE d.status='accepted' AND "+LATEST_DECISION
+
+
+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,
+        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+
+        ' ORDER BY e.year,v.acronym,e.track,s.id',(paper,))]
+
+
 def main(argv=None, *, prog=None):
     parser = argparse.ArgumentParser(prog=prog)
     parser.add_argument('--db', required=True)
@@ -15,6 +31,7 @@ def main(argv=None, *, prog=None):
     parser.add_argument('--year', type=int)
     parser.add_argument('--track',help='Exact track, e.g. main, findings, workshop')
     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)
     parser.add_argument('--stats', action='store_true')
     args = parser.parse_args(argv)
@@ -25,6 +42,10 @@ def main(argv=None, *, prog=None):
         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')]
         result['decision_events_by_status'] = [dict(r) for r in conn.execute('SELECT status,count(*) AS events FROM decisions GROUP BY status')]
+        accepted=conn.execute('SELECT count(DISTINCT paper_id) FROM ('+ACCEPTED_PAPERS+')').fetchone()[0]
+        result['papers_with_known_acceptance']=accepted
+        result['papers_without_known_acceptance']=result['papers']-accepted
+        result['outcome_scope']='Only imported evidence is known. No known acceptance is not proof that a paper was never accepted.'
         result['coverage'] = [dict(r) for r in conn.execute('SELECT status,count(*) AS sources FROM edition_sources GROUP BY status')]
         result['runs'] = [dict(r) for r in conn.execute('SELECT * FROM ingest_runs ORDER BY id DESC LIMIT 10')]
     else:
@@ -36,6 +57,8 @@ def main(argv=None, *, prog=None):
         if args.tag:
             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])
+        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:
             clause = ['1=1']
             if args.venue:
@@ -48,14 +71,7 @@ def main(argv=None, *, prog=None):
             if args.year:
                 clause.append('e.year=?'); params.append(args.year)
             if args.status:
-                latest='''EXISTS (
-                    SELECT 1 FROM decisions d WHERE d.submission_id=s.id AND d.status=?
-                    AND NOT EXISTS (
-                        SELECT 1 FROM decisions newer WHERE newer.submission_id=d.submission_id
-                        AND (newer.observed_at,COALESCE(newer.decision_date,newer.observed_at),newer.id)
-                          > (d.observed_at,COALESCE(d.decision_date,d.observed_at),d.id)
-                    )
-                )'''
+                latest='EXISTS (SELECT 1 FROM decisions d WHERE d.submission_id=s.id AND d.status=? AND '+LATEST_DECISION+')'
                 clause.append('(NOT EXISTS(SELECT 1 FROM decisions d WHERE d.submission_id=s.id) OR '+latest+')' if args.status=='unknown' else latest); params.append(args.status)
             where.append('p.id IN (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 '+' AND '.join(clause)+')')
         sql = 'SELECT p.* FROM papers p'+(' WHERE '+' AND '.join(where) if where else '')+' ORDER BY p.id LIMIT ?'
@@ -65,6 +81,8 @@ def main(argv=None, *, prog=None):
             paper = dict(row)
             paper['authors'] = json.loads(paper.pop('authors_json'))
             paper['identifiers'] = [dict(r) for r in conn.execute('SELECT scheme,value FROM identifiers WHERE paper_id=?',(row['id'],))]
+            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['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'],))]
@@ -77,6 +95,7 @@ def main(argv=None, *, prog=None):
                         if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='decision_sources'").fetchone():
                             decision['sources']=[dict(r) for r in conn.execute('SELECT s.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 s ON s.id=r.source_id WHERE ds.decision_id=?',(d['id'],))]
                         submission['decisions'].append(decision)
+                    submission['latest_status']=submission['decisions'][-1]['status'] if submission['decisions'] else 'unknown'
                     if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='submission_sources'").fetchone():
                         submission['sources']=[dict(r) for r in conn.execute('SELECT so.name,r.external_id,ss.source_submission_id FROM submission_sources ss JOIN source_records r ON r.id=ss.source_record_id JOIN sources so ON so.id=r.source_id WHERE ss.submission_id=?',(sub['id'],))]
                     submissions.append(submission)

tests/test_arxiv_audit.py

@@ -0,0 +1,44 @@
+from types import SimpleNamespace
+from unittest.mock import patch
+import json
+from pathlib import Path
+import sqlite3
+from contextlib import closing
+import tempfile
+import unittest
+import zipfile
+
+from paper_atlas.audits.arxiv import audit,MEMBER
+
+
+class ArxivSourceAuditTests(unittest.TestCase):
+    def setUp(self):
+        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup);self.root=Path(self.temp.name)
+        self.archive=self.root/'metadata.zip';self.out=self.root/'audit.json'
+        disk=patch('paper_atlas.audits.arxiv.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3));disk.start();self.addCleanup(disk.stop)
+
+    def write(self,body):
+        with zipfile.ZipFile(self.archive,'w',compression=zipfile.ZIP_DEFLATED) as z:z.writestr(MEMBER,body)
+
+    def test_repeated_ids_and_payload_changes_are_distinct_from_unique_papers(self):
+        one={'id':'2601.00001','title':'A paper','abstract':'First text'};updated={**one,'abstract':'Revised text'}
+        two={'id':'2601.00002','title':'Another paper','abstract':''}
+        self.write('\n'.join(json.dumps(row) for row in (one,updated,updated,two)))
+        result=audit(self.archive,self.out)
+        self.assertEqual((result['raw_rows'],result['unique_external_ids'],result['extra_rows_for_repeated_ids']),(4,2,2))
+        self.assertTrue(result['zip_eof_crc_verified']);self.assertEqual(result['missing_abstract_rows'],1)
+        duplicate=result['duplicate_ids'][0];self.assertEqual(duplicate['occurrences'],3)
+        self.assertTrue(duplicate['has_payload_variants']);self.assertEqual([v['occurrences'] for v in duplicate['variants']],[1,2])
+        with closing(sqlite3.connect(result['id_index'])) as c:self.assertEqual(c.execute('SELECT count(*) FROM raw_ids').fetchone()[0],2)
+        with self.assertRaisesRegex(ValueError,'output exists'):audit(self.archive,self.out)
+
+    def test_invalid_or_unverified_input_never_publishes_audit_outputs(self):
+        self.write('{"id":"2601.00001","title":"Valid"}\nnot JSON')
+        with self.assertRaises(json.JSONDecodeError):audit(self.archive,self.out)
+        self.assertFalse(self.out.exists());self.assertFalse(self.out.with_suffix('.ids.sqlite').exists())
+        self.write('{"id":"2601.00001","title":"Valid"}')
+        with self.assertRaisesRegex(ValueError,'checksum mismatch'):audit(self.archive,self.out,'0'*64)
+        self.assertFalse(self.out.exists())
+
+
+if __name__=='__main__':unittest.main()

tests/test_arxiv_tags.py

@@ -0,0 +1,49 @@
+from contextlib import closing
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.arxiv_tags import repair
+
+
+class ArxivTagRepairTests(unittest.TestCase):
+    def setUp(self):
+        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup);self.root=Path(self.temp.name)
+        self.db=self.root/'papers.sqlite';self.out=self.root/'repair.json'
+        disk=patch('paper_atlas.ingestion.arxiv_tags.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3));disk.start();self.addCleanup(disk.stop)
+        store=Store(self.db)
+        for key in ('one','two'):
+            paper,record=store.record('arxiv',key,'https://arxiv.org/abs/'+key,'Same title','AB',[],{'doi':'10.1234/same'},{'categories':'cs.AI cs.LG'})
+            for category in ('cs.AI','cs.LG'):
+                store.conn.execute("INSERT OR IGNORE INTO tags(namespace,value) VALUES('arxiv:category',?)",(category,))
+                tag=store.conn.execute("SELECT id FROM tags WHERE namespace='arxiv:category' AND value=?",(category,)).fetchone()[0]
+                store.conn.execute("INSERT OR IGNORE INTO paper_tags VALUES(?,?,?,'arxiv:categories')",(paper,tag,record))
+        tag=store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','cs.AI')").lastrowid
+        store.conn.execute("INSERT INTO paper_tags VALUES(?,?,NULL,'manual:1')",(paper,tag));store.conn.commit();store.conn.close()
+
+    def test_restore_collided_evidence_and_preserve_custom_tags_idempotently(self):
+        result=repair(self.db,self.out,batch_size=1)
+        self.assertEqual(result['records_refreshed'],2);self.assertEqual(result['status'],'complete')
+        with closing(sqlite3.connect(self.db)) as c:
+            self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],5)
+            self.assertEqual(c.execute('SELECT count(DISTINCT source_record_id) FROM paper_tags').fetchone()[0],2)
+            self.assertEqual(c.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='manual:1'").fetchone()[0],1)
+        repair(self.db,self.root/'again.json')
+        with closing(sqlite3.connect(self.db)) as c:self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],5)
+
+    def test_failed_batch_resumes_only_after_last_committed_record(self):
+        with closing(sqlite3.connect(self.db)) as c,c:c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='two'",(json.dumps({'categories':['invalid']}),))
+        with self.assertRaisesRegex(ValueError,'categories must be text'):repair(self.db,self.out,batch_size=1)
+        state=json.loads(self.out.read_text());self.assertEqual(state['records_refreshed'],1);self.assertEqual(state['status'],'failed')
+        with closing(sqlite3.connect(self.db)) as c,c:c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='two'",(json.dumps({'categories':'cs.AI cs.LG'}),))
+        result=repair(self.db,self.out,resume=True,batch_size=1)
+        self.assertEqual(result['records_refreshed'],2);self.assertEqual(result['status'],'complete')
+        with closing(sqlite3.connect(self.db)) as c:self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],5)
+
+
+if __name__=='__main__':unittest.main()

tests/test_import_arxiv.py

@@ -7,6 +7,8 @@ import zipfile
 from pathlib import Path
 
 from paper_atlas.ingestion.arxiv import ingest
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.arxiv import refresh_categories
 
 
 class ImportTests(unittest.TestCase):
@@ -53,6 +55,28 @@ class ImportTests(unittest.TestCase):
         with closing(sqlite3.connect(self.db)) as c, c:
             self.assertEqual(c.execute("SELECT count(*) FROM paper_tags pt JOIN tags t ON pt.tag_id=t.id WHERE t.value='cs.AI'").fetchone()[0],2)
 
+    def test_merged_records_retain_separate_category_evidence_across_refresh(self):
+        one=self.row('2601.00001','10.0000/example');two=self.row('2601.00002','10.0000/example')
+        self.write([one,two]);ingest(self.db,self.archive)
+        with closing(sqlite3.connect(self.db)) as c:
+            self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],4)
+            self.assertEqual(c.execute('SELECT count(DISTINCT source_record_id) FROM paper_tags').fetchone()[0],2)
+        one['categories']='cs.AI';two['categories']='cs.LG';self.write([one,two]);ingest(self.db,self.archive)
+        with closing(sqlite3.connect(self.db)) as c:
+            actual=c.execute('SELECT r.external_id,t.value FROM paper_tags pt JOIN source_records r ON r.id=pt.source_record_id JOIN tags t ON t.id=pt.tag_id ORDER BY r.external_id').fetchall()
+            self.assertEqual(actual,[('2601.00001','cs.AI'),('2601.00002','cs.LG')])
+
+    def test_migrate_legacy_categories_without_removing_custom_evidence(self):
+        store=Store(self.db);self.addCleanup(store.conn.close)
+        paper,record=store.record('arxiv','2601.00001','https://arxiv.org/abs/2601.00001','A title','AB',[])
+        for namespace,key in [('arxiv:category','arxiv:categories'),('custom:topic','manual:1')]:
+            tag=store.conn.execute('INSERT INTO tags(namespace,value) VALUES(?,?)',(namespace,'cs.AI')).lastrowid
+            store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,key))
+        refresh_categories(store,paper,record,'cs.AI cs.LG');refresh_categories(store,paper,record,'cs.AI cs.LG')
+        self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0],3)
+        self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='arxiv:categories'").fetchone()[0],0)
+        self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='manual:1'").fetchone()[0],1)
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_index_core.py

@@ -9,6 +9,17 @@ from paper_atlas.ingestion.records import ingest_file
 
 
 class StoreTests(unittest.TestCase):
+    def test_partial_source_refresh_preserves_known_abstract_and_can_later_restore_it(self):
+        paper,record=self.store.record('cvf','paper','https://example.org/paper','Known title','Known abstract',['Jane Doe'])
+        self.store.record('cvf','paper','https://example.org/paper','Known title',None,[],metadata={'abstract_fetch_error':{'status':404}})
+        self.store.record('cvf','paper','https://example.org/paper','Known title',None,[],metadata={'abstract_fetch_error':{'status':404}})
+        c=self.store.conn
+        self.assertEqual(c.execute('SELECT abstract,authors_json FROM papers').fetchone(),('Known abstract','["Jane Doe"]'))
+        source=c.execute('SELECT abstract,metadata_json FROM source_records WHERE id=?',(record,)).fetchone()
+        self.assertIsNone(source[0]);self.assertEqual(json.loads(source[1])['last_known_abstract']['abstract'],'Known abstract')
+        self.store.record('cvf','paper','https://example.org/paper','Updated title','Restored revised abstract',['Jane Doe'])
+        self.assertEqual(c.execute('SELECT title,abstract FROM papers').fetchone(),('Updated title','Restored revised abstract'))
+
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)

tests/test_query.py

@@ -9,6 +9,41 @@ from paper_atlas.db.store import Store
 
 
 class QueryTests(unittest.TestCase):
+    def run_query(self,db,*args):
+        return json.loads(subprocess.run([sys.executable,'-B','-m','paper_atlas','query','--db',str(db),*args],check=True,capture_output=True,text=True).stdout)
+
+    def test_rejection_then_acceptance_at_another_venue_preserves_both_attempts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db=Path(tmp)/'papers.sqlite';store=Store(db);self.addCleanup(store.conn.close)
+            paper,first=store.record('openreview_snapshot','forum1','https://example.org/1','Original title','Abstract',['Jane Doe'],{'arxiv':'2401.00001','openreview':'forum1'})
+            store.decision(paper,first,'ICLR',2024,'main','forum1','https://example.org/reject','rejected','Reject',identities=[('openreview','forum1')],observed_at='2024-02-01T00:00:00Z')
+            same,second=store.record('colm','forum2','https://example.org/2','Revised title','Revised abstract',['Jane Doe'],{'arxiv':'2401.00001','openreview':'forum2'})
+            store.decision(same,second,'COLM',2025,'main','forum2','https://example.org/accept','accepted','Accept',identities=[('openreview','forum2')],observed_at='2025-08-01T00:00:00Z')
+            store.conn.commit();self.assertEqual(same,paper)
+            rows=self.run_query(db,'--venue','ICLR','--status','rejected','--outcome','accepted')
+            self.assertEqual([row['id'] for row in rows],[paper])
+            self.assertEqual(rows[0]['publication_outcome'],'accepted')
+            self.assertEqual([a['venue'] for a in rows[0]['known_acceptances']],['COLM'])
+            detail=self.run_query(db,'--paper',str(paper))[0]
+            self.assertEqual([(s['acronym'],s['latest_status']) for s in detail['submissions']],[('ICLR','rejected'),('COLM','accepted')])
+            self.assertEqual(self.run_query(db,'--venue','ICLR','--status','rejected','--outcome','no_known_acceptance'),[])
+            self.assertEqual(self.run_query(db,'--stats')['papers_with_known_acceptance'],1)
+
+    def test_later_rejection_elsewhere_does_not_erase_acceptance_and_unknown_is_not_never(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db=Path(tmp)/'papers.sqlite';store=Store(db);self.addCleanup(store.conn.close)
+            paper,one=store.record('source','one','https://example.org/1','One paper','Abstract',[])
+            store.decision(paper,one,'COLM',2024,'main','one','https://example.org/accept','accepted','Accept',observed_at='2024-01-01T00:00:00Z')
+            store.decision(paper,one,'ICLR',2025,'main','two','https://example.org/reject','rejected','Reject',observed_at='2025-01-01T00:00:00Z')
+            unknown,_=store.record('arxiv','other','https://example.org/other','No observed submissions','Abstract',[])
+            store.conn.commit()
+            self.assertEqual([r['id'] for r in self.run_query(db,'--outcome','accepted')],[paper])
+            self.assertEqual([r['id'] for r in self.run_query(db,'--outcome','no_known_acceptance')],[unknown])
+            store.decision(paper,one,'COLM',2024,'main','one','https://example.org/withdraw','withdrawn','Withdrawn',observed_at='2026-01-01T00:00:00Z')
+            store.conn.commit();self.assertEqual(self.run_query(db,'--outcome','accepted'),[])
+            history=self.run_query(db,'--paper',str(paper))[0]
+            self.assertEqual([d['status'] for d in history['submissions'][0]['decisions']],['accepted','withdrawn'])
+
     def test_joint_conference_alias_does_not_duplicate_submission(self):
         with tempfile.TemporaryDirectory() as tmp:
             db=Path(tmp)/'papers.sqlite';store=Store(db);self.addCleanup(store.conn.close)

tests/test_vision.py

@@ -1,11 +1,85 @@
 import tempfile
+import io
+import json
+from pathlib import Path
 import unittest
 from unittest.mock import patch
+from urllib.error import HTTPError
 
-from paper_atlas.collectors.vision import collect_index, decode_html, discover, index_entries, parse_paper
+from paper_atlas.collectors.vision import collect_index, decode_html, discover, index_entries, parse_paper,main
 
 
 class VisionTests(unittest.TestCase):
+    def cvf_fixture(self,year=2018):
+        root=f'https://openaccess.thecvf.com/CVPR{year}.py'
+        papers=[f'https://openaccess.thecvf.com/content_cvpr_{year}/html/Paper{i}_CVPR_{year}_paper.html' for i in (1,2,3)]
+        def entry(url,title='Paper'):
+            return f'<dt><a href="{url}">{title}</a></dt><dd>Jane Doe, John Smith</dd>'
+        return root,papers,entry
+
+    def test_all_explicit_days_are_combined_and_overlapping_papers_are_not_duplicated(self):
+        root,papers,entry=self.cvf_fixture();day1=root+'?day=2018-06-19';day2=root+'?day=2018-06-20'
+        pages={root:f'<a href="{day1}">Day1</a><a href="{day2}">Day2</a><a href="https://evil.example/CVPR2018.py?day=all">Other</a><a href="/CVPR2019.py?day=all">Wrong year</a>',day1:'<dl>'+entry(papers[0])+entry(papers[1])+'</dl>',day2:'<dl>'+entry(papers[1])+entry(papers[2])+'</dl>'}
+        pages.update({paper:'<div id="papertitle">Paper</div><div id="abstract">Abstract</div>' for paper in papers})
+        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)
+            rows=[json.loads(line) for line in Path(manifest['file']).read_text().splitlines()]
+        self.assertEqual(manifest['count'],3);self.assertEqual(manifest['raw_index_entry_count'],4)
+        self.assertEqual(manifest['index_urls'],[day1,day2]);self.assertEqual(manifest['status'],'collected')
+        self.assertEqual(rows[1]['metadata']['index_urls'],[day1,day2])
+        self.assertEqual({call.args[0] for call in fetch.call_args_list},set(pages))
+
+    def test_conflicting_copies_across_days_fail_before_publishing(self):
+        root,papers,entry=self.cvf_fixture();days=[root+'?day=2018-06-19',root+'?day=2018-06-20']
+        pages={root:''.join(f'<a href="{day}">Day</a>' for day in days),days[0]:'<dl>'+entry(papers[0],'First title')+'</dl>',days[1]:'<dl>'+entry(papers[0],'Different title')+'</dl>'}
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=lambda url,*a,**kw:pages[url].encode()):
+            with self.assertRaisesRegex(ValueError,'Conflicting metadata'):
+                collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            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):
+            if url==root:return ('<dl>'+entry(papers[0],'Known publication')+'</dl>').encode()
+            raise error
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=fetch):
+            manifest=collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            row=json.loads(Path(manifest['file']).read_text())
+        self.assertEqual(row['title'],'Known publication');self.assertIsNone(row['abstract'])
+        self.assertEqual(row['authors'],['Jane Doe','John Smith']);self.assertEqual(row['evidence_url'],root)
+        self.assertEqual(manifest['status'],'partial');self.assertEqual(manifest['missing_abstract_pages'],[papers[0]])
+        self.assertEqual(row['metadata']['abstract_fetch_error']['status'],404)
+
+    def test_forbidden_paper_is_not_replaced_with_a_fallback(self):
+        root,papers,entry=self.cvf_fixture()
+        def fetch(url,*args,**kwargs):
+            if url==root:return ('<dl>'+entry(papers[0])+'</dl>').encode()
+            raise HTTPError(url,403,'Forbidden',{},io.BytesIO())
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=fetch):
+            with self.assertRaises(HTTPError):collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            self.assertFalse(list(Path(tmp).glob('*.jsonl')))
+
+    def test_broken_all_papers_link_can_use_all_advertised_days(self):
+        root,papers,entry=self.cvf_fixture();day=root+'?day=2018-06-19';all_url=root+'?day=all'
+        pages={root:f'<a href="{day}">Day</a><a href="{all_url}">All</a>',day:'<dl>'+entry(papers[0])+'</dl>',papers[0]:'<div id="papertitle">Paper</div><div id="abstract">Abstract</div>'}
+        def fetch(url,*a,**kw):
+            if url==all_url:raise HTTPError(url,404,'Not Found',{},io.BytesIO())
+            return pages[url].encode()
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=fetch):
+            manifest=collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+        self.assertEqual(manifest['index_urls'],[day]);self.assertEqual(manifest['index_errors'],[{'url':all_url,'status':404}])
+
+    def test_selective_retry_uses_separate_output_and_leaves_original_catalogue_intact(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            old=Path(tmp)/'old.json';payload={'CVPR:2018:main':{'status':'blocked'},'CVPR:2019:main':{'status':'collected'}};old.write_text(json.dumps(payload))
+            indexes=[{'venue':'CVPR','year':year,'track':'main','index_url':f'https://openaccess.thecvf.com/CVPR{year}'} for year in (2018,2019)]
+            output=Path(tmp)/'repaired'
+            with patch('paper_atlas.collectors.vision.discover',return_value=indexes),patch('paper_atlas.collectors.vision.collect_index',return_value={'status':'collected','count':1}) as collect:
+                main(['--cache',tmp,'--out',str(output),'--retry-catalogue',str(old)],source='cvf')
+            self.assertEqual(collect.call_count,1);self.assertEqual(collect.call_args.args[1]['year'],2018)
+            self.assertEqual(json.loads(old.read_text()),payload)
+            self.assertEqual(set(json.loads((output/'cvf-index-status.json').read_text())),{'CVPR:2018:main'})
+
     def entry(self):
         return {'url':'https://www.ecva.net/papers/eccv_2024/papers_ECCV/html/1_ECCV_2024_paper.php','index_title':'Paper','index_authors':'Jane Doe*, John Smith','doi':'10.1007/example_1'}