feat(literature-search-verify): 新增 OpenAlex 检索源、限流重试与单元测试

新增 scripts/http_utils.py 统一封装 HTTP 请求的限流重试(429/5xx 指数退避,
优先遵守服务端 Retry-After),search_arxiv/crossref/semantic_scholar.py 和
verify_citation.py 都已接入,不再各自裸调 urllib。

新增 scripts/search_openalex.py 作为第四个检索源:免费、无需 API key,覆盖面
比单独的 Crossref 更广,还能拿到开放获取PDF直链;已接入 literature_search.py
的主检索流程。verify_citation.py 的跨源标题核查同步改为同时查 Semantic
Scholar 和 OpenAlex 两个独立源、任一命中相似度达标即通过,不再单点依赖 S2——
这是针对"S2 被限流导致整批候选退化成 unverified"这个实际发生过的问题的直接
修复,已用真实网络请求验证:复测中 S2 确实当场返回了 429,靠 OpenAlex 兜底
最终判定仍然是 verified。

顺带修了 archive_references.py 的 slugify(),之前中文主题名会被正则全部
过滤掉、退化成通用的 "references",导致不同中文主题的归档目录互相冲突。

scripts/tests/ 下补了 37 个 unittest(全部 mock 网络请求,不发真实请求),
覆盖 verify_citation 的三档判定和双源核查合并逻辑、archive_references 的
bib 解析边界情况(嵌套花括号、中文主题名)、literature_search 的候选去重
合并与 BibTeX 生成、http_utils 的重试逻辑。用标准库 unittest 而不是 pytest,
和这些脚本本身不引入第三方依赖的原则保持一致。

CLAUDE.md 的下一步计划里,这一项已标记为完成。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.claude/skills/literature-search-verify/scripts/archive_references.py

@@ -35,8 +35,12 @@ from datetime import date
 
 
 def slugify(text):
+    """ASCII gets lowercased; CJK characters are kept as-is (there's no
+    reversible case-fold for them) so an all-Chinese topic name doesn't
+    collapse to nothing and fall back to the generic "references" name,
+    colliding with every other Chinese-named topic archived in this repo."""
     text = text.strip().lower()
-    text = re.sub(r"[^a-z0-9]+", "_", text)
+    text = re.sub(r"[^a-z0-9一-鿿]+", "_", text)
     return text.strip("_")[:60] or "references"
 
 

.claude/skills/literature-search-verify/scripts/http_utils.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""
+Shared HTTP GET helper with exponential-backoff retry for rate limiting (429)
+and transient server errors (5xx). Stdlib only.
+
+Every search_*.py / verify_citation.py network call should go through this
+instead of calling urllib.request.urlopen directly, so a single source being
+temporarily rate-limited doesn't quietly degrade a whole batch of candidates
+to "unverified" the way Semantic Scholar's 429s did before this existed.
+
+Importable:
+    from http_utils import fetch
+"""
+import time
+import urllib.request
+import urllib.error
+
+RETRYABLE_CODES = {429, 500, 502, 503, 504}
+
+
+def _wait_seconds(http_error, attempt, base=1.0, cap=20.0):
+    retry_after = http_error.headers.get("Retry-After") if http_error.headers else None
+    if retry_after:
+        try:
+            return min(float(retry_after), cap)
+        except ValueError:
+            pass
+    return min(base * (2 ** attempt), cap)
+
+
+def fetch(url, headers=None, timeout=20, max_retries=4):
+    """GET url, retrying on 429/5xx with exponential backoff (honoring a
+    Retry-After header when the server sends one). Returns the raw response
+    bytes. Raises the underlying urllib error if retries are exhausted or the
+    error isn't retryable (e.g. 404)."""
+    req = urllib.request.Request(url, headers=headers or {})
+    attempt = 0
+    while True:
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                return resp.read()
+        except urllib.error.HTTPError as e:
+            if e.code not in RETRYABLE_CODES or attempt >= max_retries:
+                raise
+            time.sleep(_wait_seconds(e, attempt))
+            attempt += 1

.claude/skills/literature-search-verify/scripts/literature_search.py

@@ -1,8 +1,9 @@
 #!/usr/bin/env python3
 """
-End-to-end literature search: query arXiv + Semantic Scholar + Crossref,
-merge/dedupe candidates, independently verify each one, and emit both a
-human-readable report and BibTeX for the entries that passed verification.
+End-to-end literature search: query arXiv + Semantic Scholar + Crossref +
+OpenAlex, merge/dedupe candidates, independently verify each one, and emit
+both a human-readable report and BibTeX for the entries that passed
+verification.
 
 This is the one script Claude should actually call for a normal literature
 search -- the individual search_*.py / verify_citation.py scripts exist
@@ -28,6 +29,7 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from search_arxiv import search_arxiv
 from search_semantic_scholar import search_s2
 from search_crossref import search_crossref
+from search_openalex import search_openalex
 from verify_citation import verify
 
 
@@ -117,7 +119,13 @@ def to_bibtex(candidate, key):
 def run(query, max_per_source=8):
     all_results = []
     errors = {}
-    for name, fn in (("arxiv", search_arxiv), ("semantic_scholar", search_s2), ("crossref", search_crossref)):
+    sources = (
+        ("arxiv", search_arxiv),
+        ("semantic_scholar", search_s2),
+        ("crossref", search_crossref),
+        ("openalex", search_openalex),
+    )
+    for name, fn in sources:
         try:
             all_results.extend(fn(query, max_per_source))
         except Exception as e:

.claude/skills/literature-search-verify/scripts/search_arxiv.py

@@ -9,12 +9,15 @@ Importable:
     from search_arxiv import search_arxiv
 """
 import sys
+import os
 import json
 import argparse
-import urllib.request
 import urllib.parse
 import xml.etree.ElementTree as ET
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from http_utils import fetch
+
 ARXIV_API = "http://export.arxiv.org/api/query"
 NS = {"atom": "http://www.w3.org/2005/Atom"}
 
@@ -28,8 +31,7 @@ def search_arxiv(query, max_results=10, timeout=20):
         "sortOrder": "descending",
     }
     url = f"{ARXIV_API}?{urllib.parse.urlencode(params)}"
-    with urllib.request.urlopen(url, timeout=timeout) as resp:
-        data = resp.read()
+    data = fetch(url, timeout=timeout)
     root = ET.fromstring(data)
     results = []
     for entry in root.findall("atom:entry", NS):

.claude/skills/literature-search-verify/scripts/search_crossref.py

@@ -10,11 +10,14 @@ Importable:
     from search_crossref import search_crossref
 """
 import sys
+import os
 import json
 import argparse
-import urllib.request
 import urllib.parse
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from http_utils import fetch
+
 CROSSREF_API = "https://api.crossref.org/works"
 UA = "literature-search-verify-skill/1.0 (mailto:research-assistant@example.com)"
 
@@ -22,9 +25,7 @@ UA = "literature-search-verify-skill/1.0 (mailto:research-assistant@example.com)
 def search_crossref(query, max_results=10, timeout=20):
     params = {"query": query, "rows": max_results}
     url = f"{CROSSREF_API}?{urllib.parse.urlencode(params)}"
-    req = urllib.request.Request(url, headers={"User-Agent": UA})
-    with urllib.request.urlopen(req, timeout=timeout) as resp:
-        data = json.loads(resp.read())
+    data = json.loads(fetch(url, headers={"User-Agent": UA}, timeout=timeout))
     results = []
     for item in data.get("message", {}).get("items", []) or []:
         titles = item.get("title") or []

.claude/skills/literature-search-verify/scripts/search_openalex.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""
+Search the OpenAlex Works API. No API key required -- set the OPENALEX_MAILTO
+environment variable to an email address to use OpenAlex's "polite pool" for
+higher and more reliable rate limits (OpenAlex's own recommendation, not a
+credential). Coverage is broader than Crossref alone (includes arXiv, IEEE,
+and most publishers' metadata in one place) and each result carries an
+open-access PDF link when one exists.
+
+CLI usage:
+    python3 search_openalex.py "UAV magnetic compensation" --max 10
+
+Importable:
+    from search_openalex import search_openalex
+"""
+import sys
+import os
+import json
+import argparse
+import urllib.parse
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from http_utils import fetch
+
+OPENALEX_API = "https://api.openalex.org/works"
+SELECT_FIELDS = ",".join([
+    "doi", "title", "display_name", "publication_year", "authorships",
+    "primary_location", "open_access", "cited_by_count", "abstract_inverted_index",
+])
+
+
+def _reconstruct_abstract(inverted_index):
+    """OpenAlex returns abstracts as a word -> [positions] inverted index
+    (for copyright reasons) instead of plain text; rebuild the plain text."""
+    if not inverted_index:
+        return None
+    positions = {}
+    for word, idxs in inverted_index.items():
+        for i in idxs:
+            positions[i] = word
+    if not positions:
+        return None
+    return " ".join(positions[i] for i in sorted(positions))
+
+
+def search_openalex(query, max_results=10, timeout=20):
+    params = {"search": query, "per_page": max_results, "select": SELECT_FIELDS}
+    mailto = os.environ.get("OPENALEX_MAILTO")
+    if mailto:
+        params["mailto"] = mailto
+    url = f"{OPENALEX_API}?{urllib.parse.urlencode(params)}"
+    data = json.loads(fetch(url, timeout=timeout))
+    results = []
+    for item in data.get("results", []) or []:
+        title = item.get("title") or item.get("display_name")
+        authors = [
+            a.get("author", {}).get("display_name")
+            for a in (item.get("authorships") or [])
+            if a.get("author")
+        ]
+        doi = item.get("doi")
+        if doi and doi.startswith("https://doi.org/"):
+            doi = doi[len("https://doi.org/"):]
+        primary_location = item.get("primary_location") or {}
+        source = primary_location.get("source") or {}
+        open_access = item.get("open_access") or {}
+        results.append({
+            "source": "openalex",
+            "title": title,
+            "authors": authors,
+            "year": item.get("publication_year"),
+            "venue": source.get("display_name"),
+            "doi": doi,
+            "arxiv_id": None,
+            "abstract": _reconstruct_abstract(item.get("abstract_inverted_index")),
+            "citation_count": item.get("cited_by_count"),
+            "pdf_url": open_access.get("oa_url") or primary_location.get("pdf_url"),
+        })
+    return results
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("query")
+    ap.add_argument("--max", type=int, default=10)
+    args = ap.parse_args()
+    try:
+        out = search_openalex(args.query, args.max)
+        print(json.dumps(out, ensure_ascii=False, indent=2))
+    except Exception as e:
+        print(json.dumps({"error": str(e)}, ensure_ascii=False))
+        sys.exit(1)

.claude/skills/literature-search-verify/scripts/search_semantic_scholar.py

@@ -13,9 +13,11 @@ import sys
 import os
 import json
 import argparse
-import urllib.request
 import urllib.parse
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from http_utils import fetch
+
 S2_API = "https://api.semanticscholar.org/graph/v1/paper/search"
 FIELDS = "title,authors,year,venue,externalIds,abstract,citationCount"
 
@@ -23,12 +25,11 @@ FIELDS = "title,authors,year,venue,externalIds,abstract,citationCount"
 def search_s2(query, max_results=10, timeout=20):
     params = {"query": query, "limit": max_results, "fields": FIELDS}
     url = f"{S2_API}?{urllib.parse.urlencode(params)}"
-    req = urllib.request.Request(url)
+    headers = {}
     api_key = os.environ.get("S2_API_KEY")
     if api_key:
-        req.add_header("x-api-key", api_key)
-    with urllib.request.urlopen(req, timeout=timeout) as resp:
-        data = json.loads(resp.read())
+        headers["x-api-key"] = api_key
+    data = json.loads(fetch(url, headers=headers, timeout=timeout))
     results = []
     for p in data.get("data", []) or []:
         ext = p.get("externalIds") or {}

.claude/skills/literature-search-verify/scripts/tests/test_archive_references.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""
+Unit tests for archive_references.py's slugify() and the minimal bib parser,
+covering the edge cases that are easy to get wrong silently: CJK topic names
+and nested-brace LaTeX field values (e.g. \\ensuremath{\\epsilon}).
+
+Run: python -m unittest discover -s .claude/skills/literature-search-verify/scripts
+"""
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
+import archive_references as ar
+
+
+class TestSlugify(unittest.TestCase):
+    def test_ascii_topic(self):
+        self.assertEqual(ar.slugify("UAV aeromagnetic compensation"), "uav_aeromagnetic_compensation")
+
+    def test_chinese_topic_is_preserved_not_collapsed_to_generic_name(self):
+        slug = ar.slugify("无人机磁补偿")
+        self.assertNotEqual(slug, "references")
+        self.assertIn("补偿", slug)
+
+    def test_mixed_chinese_and_english(self):
+        slug = ar.slugify("无人机磁补偿 Tolles-Lawson")
+        self.assertTrue(slug.endswith("tolles_lawson"))
+
+    def test_blank_topic_falls_back_to_references(self):
+        self.assertEqual(ar.slugify("   "), "references")
+
+    def test_truncated_to_60_chars(self):
+        self.assertEqual(len(ar.slugify("a" * 100)), 60)
+
+
+class TestParseBibEntries(unittest.TestCase):
+    def _write_bib(self, content):
+        fd, path = tempfile.mkstemp(suffix=".bib")
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(content)
+        self.addCleanup(os.remove, path)
+        return path
+
+    def test_nested_braces_in_field_value_do_not_truncate_it(self):
+        bib = (
+            "@article{wu2017,\n"
+            "  title   = {Aeromagnetic gradient compensation using \\ensuremath{\\epsilon}-SVR},\n"
+            "  journal = {Journal of Applied Remote Sensing},\n"
+            "  year    = {2017},\n"
+            "  doi     = {10.1117/1.jrs.11.025012},\n"
+            "}"
+        )
+        entries = ar.parse_bib_entries(self._write_bib(bib))
+        self.assertEqual(len(entries), 1)
+        self.assertIn("\\ensuremath{\\epsilon}-SVR", entries[0]["title"])
+        self.assertEqual(entries[0]["year"], "2017")
+
+    def test_multiple_entries_and_optional_fields(self):
+        bib = (
+            "@article{a2020,\n"
+            "  title = {First paper},\n"
+            "  year  = {2020},\n"
+            "}\n\n"
+            "@article{b2021,\n"
+            "  title = {Second paper},\n"
+            "  year  = {2021},\n"
+            "  note  = {some note},\n"
+            "}"
+        )
+        entries = ar.parse_bib_entries(self._write_bib(bib))
+        self.assertEqual([e["key"] for e in entries], ["a2020", "b2021"])
+        self.assertNotIn("note", entries[0])
+        self.assertEqual(entries[1]["note"], "some note")
+
+
+class TestYearSortKey(unittest.TestCase):
+    def test_entries_without_a_parseable_year_sort_last(self):
+        entries = [{"key": "b", "year": ""}, {"key": "a", "year": "1999"}]
+        entries.sort(key=ar.year_sort_key)
+        self.assertEqual([e["key"] for e in entries], ["a", "b"])
+
+
+class TestBuildReadme(unittest.TestCase):
+    def test_omits_optional_sections_when_not_given(self):
+        readme = ar.build_readme("Topic", [], 0, [], None)
+        self.assertNotIn("Flagged during search", readme)
+        self.assertNotIn("Search coverage notes", readme)
+
+    def test_includes_suspect_and_notes_when_given(self):
+        readme = ar.build_readme("Topic", [], 0, ["Bad title|dubious venue"], "coverage notes here")
+        self.assertIn("Bad title", readme)
+        self.assertIn("dubious venue", readme)
+        self.assertIn("coverage notes here", readme)
+
+
+if __name__ == "__main__":
+    unittest.main()

.claude/skills/literature-search-verify/scripts/tests/test_http_utils.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""
+Unit tests for http_utils.fetch's retry/backoff behavior. All network calls
+are mocked -- these never touch the real network.
+
+Run: python -m unittest discover -s .claude/skills/literature-search-verify/scripts
+"""
+import os
+import sys
+import unittest
+import urllib.error
+import urllib.request
+from email.message import Message
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
+import http_utils
+
+
+def _http_error(code, retry_after=None):
+    hdrs = Message()
+    if retry_after is not None:
+        hdrs["Retry-After"] = str(retry_after)
+    return urllib.error.HTTPError("http://example.com", code, "err", hdrs, None)
+
+
+class FakeResponse:
+    def __init__(self, data):
+        self._data = data
+
+    def read(self):
+        return self._data
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        return False
+
+
+class TestFetch(unittest.TestCase):
+    @mock.patch("urllib.request.urlopen")
+    @mock.patch("http_utils.time.sleep", return_value=None)
+    def test_succeeds_first_try_without_sleeping(self, mock_sleep, mock_urlopen):
+        mock_urlopen.return_value = FakeResponse(b"ok")
+        result = http_utils.fetch("http://example.com")
+        self.assertEqual(result, b"ok")
+        mock_sleep.assert_not_called()
+
+    @mock.patch("urllib.request.urlopen")
+    @mock.patch("http_utils.time.sleep", return_value=None)
+    def test_retries_on_429_then_succeeds(self, mock_sleep, mock_urlopen):
+        mock_urlopen.side_effect = [_http_error(429), FakeResponse(b"ok")]
+        result = http_utils.fetch("http://example.com", max_retries=3)
+        self.assertEqual(result, b"ok")
+        self.assertEqual(mock_urlopen.call_count, 2)
+        mock_sleep.assert_called_once()
+
+    @mock.patch("urllib.request.urlopen")
+    @mock.patch("http_utils.time.sleep", return_value=None)
+    def test_retries_on_5xx(self, mock_sleep, mock_urlopen):
+        mock_urlopen.side_effect = [_http_error(503), FakeResponse(b"ok")]
+        result = http_utils.fetch("http://example.com", max_retries=3)
+        self.assertEqual(result, b"ok")
+
+    @mock.patch("urllib.request.urlopen")
+    @mock.patch("http_utils.time.sleep", return_value=None)
+    def test_raises_after_exhausting_retries(self, mock_sleep, mock_urlopen):
+        mock_urlopen.side_effect = [_http_error(429)] * 10
+        with self.assertRaises(urllib.error.HTTPError):
+            http_utils.fetch("http://example.com", max_retries=2)
+        self.assertEqual(mock_urlopen.call_count, 3)  # initial attempt + 2 retries
+
+    @mock.patch("urllib.request.urlopen")
+    @mock.patch("http_utils.time.sleep", return_value=None)
+    def test_non_retryable_status_raises_immediately(self, mock_sleep, mock_urlopen):
+        mock_urlopen.side_effect = _http_error(404)
+        with self.assertRaises(urllib.error.HTTPError):
+            http_utils.fetch("http://example.com")
+        self.assertEqual(mock_urlopen.call_count, 1)
+        mock_sleep.assert_not_called()
+
+
+class TestWaitSeconds(unittest.TestCase):
+    def test_honors_retry_after_header(self):
+        err = _http_error(429, retry_after=5)
+        self.assertEqual(http_utils._wait_seconds(err, attempt=0), 5.0)
+
+    def test_falls_back_to_exponential_backoff_without_header(self):
+        err = _http_error(429)
+        self.assertAlmostEqual(http_utils._wait_seconds(err, attempt=2, base=1.0), 4.0)
+
+    def test_caps_wait_time(self):
+        err = _http_error(429, retry_after=999)
+        self.assertEqual(http_utils._wait_seconds(err, attempt=0, cap=20.0), 20.0)
+
+    def test_ignores_unparseable_retry_after(self):
+        err = _http_error(429, retry_after="not-a-number")
+        self.assertAlmostEqual(http_utils._wait_seconds(err, attempt=0, base=1.0), 1.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

.claude/skills/literature-search-verify/scripts/tests/test_literature_search.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""
+Unit tests for literature_search.py's candidate merging and BibTeX generation.
+
+Run: python -m unittest discover -s .claude/skills/literature-search-verify/scripts
+"""
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
+import literature_search as ls
+
+
+class TestMergeCandidates(unittest.TestCase):
+    def test_dedupes_same_paper_across_sources_and_fills_missing_fields(self):
+        items = [
+            {"source": "arxiv", "title": "Deep Learning for Aeromagnetic Compensation",
+             "arxiv_id": "2401.00001", "doi": None, "year": "2024"},
+            {"source": "crossref", "title": "Deep Learning for Aeromagnetic Compensation",
+             "arxiv_id": None, "doi": "10.1/x", "year": 2024, "venue": "Some Journal"},
+        ]
+        merged = ls.merge_candidates(items)
+        self.assertEqual(len(merged), 1)
+        self.assertEqual(merged[0]["arxiv_id"], "2401.00001")
+        self.assertEqual(merged[0]["doi"], "10.1/x")
+        self.assertEqual(sorted(merged[0]["sources"]), ["arxiv", "crossref"])
+
+    def test_distinct_titles_are_not_merged(self):
+        items = [
+            {"source": "arxiv", "title": "Paper About Cats"},
+            {"source": "crossref", "title": "Paper About Dogs"},
+        ]
+        self.assertEqual(len(ls.merge_candidates(items)), 2)
+
+
+class TestBibtexKey(unittest.TestCase):
+    def test_collision_gets_letter_suffix(self):
+        used = set()
+        key1 = ls.make_bibtex_key({"authors": ["Jane Smith"], "year": "2020"}, used)
+        key2 = ls.make_bibtex_key({"authors": ["John Smith"], "year": "2020"}, used)
+        self.assertEqual(key1, "smith2020")
+        self.assertEqual(key2, "smith2020a")
+
+    def test_no_authors_falls_back_to_unknown(self):
+        key = ls.make_bibtex_key({"authors": [], "year": "2020"}, set())
+        self.assertEqual(key, "unknown2020")
+
+
+class TestToBibtex(unittest.TestCase):
+    def test_arxiv_only_candidate_uses_misc_with_eprint(self):
+        cand = {"authors": ["A B"], "title": "T", "year": "2024",
+                "arxiv_id": "2401.00001", "venue": None, "doi": None}
+        bib = ls.to_bibtex(cand, "b2024")
+        self.assertTrue(bib.startswith("@misc{b2024,"))
+        self.assertIn("eprint = {2401.00001}", bib)
+
+    def test_candidate_with_venue_uses_article_with_doi(self):
+        cand = {"authors": ["A B"], "title": "T", "year": "2024",
+                "arxiv_id": None, "venue": "Some Journal", "doi": "10.1/x"}
+        bib = ls.to_bibtex(cand, "b2024")
+        self.assertTrue(bib.startswith("@article{b2024,"))
+        self.assertIn("doi = {10.1/x}", bib)
+
+
+if __name__ == "__main__":
+    unittest.main()

.claude/skills/literature-search-verify/scripts/tests/test_verify_citation.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""
+Unit tests for verify_citation's verdict aggregation logic -- this is the
+anti-hallucination guarantee, so its three-way verified/suspect/unverified
+decision needs to stay correct independent of any real network call. All
+individual check_* functions are mocked; only the aggregation logic itself
+(verify() and _combine_title_checks()) is under test.
+
+Run: python -m unittest discover -s .claude/skills/literature-search-verify/scripts
+"""
+import os
+import sys
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
+import verify_citation as vc
+
+
+class TestCombineTitleChecks(unittest.TestCase):
+    """Two independent title sources (S2 + OpenAlex) shouldn't let one
+    source's incomplete coverage outvote the other's confirmation."""
+
+    def test_both_pass_picks_higher_similarity(self):
+        s2 = {"status": "pass", "similarity": 0.95, "matched_title": "A"}
+        oa = {"status": "pass", "similarity": 0.99, "matched_title": "B"}
+        combined = vc._combine_title_checks(s2, oa)
+        self.assertEqual(combined["status"], "pass")
+        self.assertEqual(combined["matched_title"], "B")
+
+    def test_one_pass_one_fail_still_passes(self):
+        s2 = {"status": "fail", "reason": "no matching title found on Semantic Scholar"}
+        oa = {"status": "pass", "similarity": 0.95, "matched_title": "B"}
+        self.assertEqual(vc._combine_title_checks(s2, oa)["status"], "pass")
+
+    def test_one_pass_one_skipped_still_passes(self):
+        s2 = {"status": "skipped", "reason": "timeout"}
+        oa = {"status": "pass", "similarity": 0.95, "matched_title": "B"}
+        self.assertEqual(vc._combine_title_checks(s2, oa)["status"], "pass")
+
+    def test_both_fail_is_fail(self):
+        s2 = {"status": "fail", "reason": "x"}
+        oa = {"status": "fail", "reason": "y"}
+        self.assertEqual(vc._combine_title_checks(s2, oa)["status"], "fail")
+
+    def test_fail_plus_skipped_is_still_fail(self):
+        s2 = {"status": "fail", "reason": "x"}
+        oa = {"status": "skipped", "reason": "y"}
+        self.assertEqual(vc._combine_title_checks(s2, oa)["status"], "fail")
+
+    def test_both_skipped_is_skipped_not_fail(self):
+        s2 = {"status": "skipped", "reason": "x"}
+        oa = {"status": "skipped", "reason": "y"}
+        self.assertEqual(vc._combine_title_checks(s2, oa)["status"], "skipped")
+
+
+class TestVerifyVerdict(unittest.TestCase):
+    def test_all_checks_pass_is_verified(self):
+        with mock.patch.object(vc, "check_arxiv_id", return_value={"status": "pass"}), \
+             mock.patch.object(vc, "check_doi", return_value={"status": "pass"}), \
+             mock.patch.object(vc, "check_title_cross_source_s2", return_value={"status": "pass", "similarity": 1.0}), \
+             mock.patch.object(vc, "check_title_cross_source_openalex", return_value={"status": "pass", "similarity": 1.0}):
+            result = vc.verify(title="T", arxiv_id="1234.5678", doi="10.1/x")
+        self.assertEqual(result["verdict"], "verified")
+
+    def test_one_failing_check_is_suspect_even_if_another_passes(self):
+        with mock.patch.object(vc, "check_arxiv_id", return_value={"status": "fail", "reason": "arXiv ID not found"}), \
+             mock.patch.object(vc, "check_doi", return_value={"status": "pass"}):
+            result = vc.verify(arxiv_id="9999.99999", doi="10.1/x")
+        self.assertEqual(result["verdict"], "suspect")
+
+    def test_all_checks_skipped_is_unverified_not_verified(self):
+        with mock.patch.object(vc, "check_arxiv_id", return_value={"status": "skipped", "reason": "timeout"}):
+            result = vc.verify(arxiv_id="1234.5678")
+        self.assertEqual(result["verdict"], "unverified")
+
+    def test_no_identifiers_given_produces_no_checks_and_unverified(self):
+        result = vc.verify(title=None, arxiv_id=None, doi=None)
+        self.assertEqual(result["checks"], {})
+        self.assertEqual(result["verdict"], "unverified")
+
+
+class TestTitleSimilarity(unittest.TestCase):
+    def test_identical_titles_score_one(self):
+        self.assertEqual(vc.title_similarity("Same Title", "same title"), 1.0)
+
+    def test_empty_or_missing_inputs_score_zero(self):
+        self.assertEqual(vc.title_similarity("", "x"), 0.0)
+        self.assertEqual(vc.title_similarity(None, "x"), 0.0)
+        self.assertEqual(vc.title_similarity("x", None), 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

.claude/skills/literature-search-verify/scripts/verify_citation.py

@@ -13,14 +13,17 @@ Importable:
     from verify_citation import verify
 """
 import sys
+import os
 import json
 import argparse
 import difflib
-import urllib.request
 import urllib.parse
 import urllib.error
 import xml.etree.ElementTree as ET
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from http_utils import fetch
+
 ATOM_NS = {"atom": "http://www.w3.org/2005/Atom"}
 
 
@@ -34,8 +37,7 @@ def check_arxiv_id(arxiv_id, timeout=20):
     """Confirm an arXiv ID actually resolves to a real paper."""
     try:
         url = f"http://export.arxiv.org/api/query?id_list={urllib.parse.quote(arxiv_id)}"
-        with urllib.request.urlopen(url, timeout=timeout) as resp:
-            data = resp.read()
+        data = fetch(url, timeout=timeout)
         root = ET.fromstring(data)
         entry = root.find("atom:entry", ATOM_NS)
         if entry is None:
@@ -51,11 +53,7 @@ def check_doi(doi, timeout=20):
     """Confirm a DOI actually resolves via Crossref."""
     try:
         url = f"https://api.crossref.org/works/{urllib.parse.quote(doi)}"
-        req = urllib.request.Request(
-            url, headers={"User-Agent": "literature-search-verify-skill/1.0"}
-        )
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
-            data = json.loads(resp.read())
+        data = json.loads(fetch(url, headers={"User-Agent": "literature-search-verify-skill/1.0"}, timeout=timeout))
         titles = data.get("message", {}).get("title") or []
         return {"status": "pass", "canonical_title": titles[0] if titles else None}
     except urllib.error.HTTPError as e:
@@ -66,27 +64,70 @@ def check_doi(doi, timeout=20):
         return {"status": "skipped", "reason": str(e)}
 
 
-def check_title_cross_source(title, timeout=20):
-    """Independently re-search by title on a different source (Semantic
-    Scholar) and require a near-exact title match. This is what catches a
-    plausible-sounding but entirely invented title/author combination."""
+def _best_title_match(candidates, title, title_field="title"):
+    best = max(candidates, key=lambda p: title_similarity(title, p.get(title_field, "") or ""))
+    sim = title_similarity(title, best.get(title_field, "") or "")
+    return sim, best.get(title_field)
+
+
+def check_title_cross_source_s2(title, timeout=20):
+    """Independently re-search by title on Semantic Scholar and require a
+    near-exact title match. This is what catches a plausible-sounding but
+    entirely invented title/author combination."""
     try:
         params = {"query": title, "limit": 3, "fields": "title"}
         url = f"https://api.semanticscholar.org/graph/v1/paper/search?{urllib.parse.urlencode(params)}"
-        with urllib.request.urlopen(url, timeout=timeout) as resp:
-            data = json.loads(resp.read())
+        data = json.loads(fetch(url, timeout=timeout))
         candidates = data.get("data", []) or []
         if not candidates:
             return {"status": "fail", "reason": "no matching title found on Semantic Scholar"}
-        best = max(candidates, key=lambda p: title_similarity(title, p.get("title", "")))
-        sim = title_similarity(title, best.get("title", ""))
-        if sim >= 0.9:
-            return {"status": "pass", "similarity": round(sim, 3), "matched_title": best.get("title")}
-        return {"status": "fail", "similarity": round(sim, 3), "matched_title": best.get("title")}
+        sim, matched = _best_title_match(candidates, title)
+        status = "pass" if sim >= 0.9 else "fail"
+        return {"status": status, "similarity": round(sim, 3), "matched_title": matched}
     except Exception as e:
         return {"status": "skipped", "reason": str(e)}
 
 
+def check_title_cross_source_openalex(title, timeout=20):
+    """Same idea as check_title_cross_source_s2 but against OpenAlex -- a
+    second, independent source for the title cross-check so a single source
+    (S2 in particular has real-world rate-limit issues) being unavailable
+    doesn't drop this whole check to 'skipped' for a batch of candidates."""
+    try:
+        params = {"search": title, "per_page": 3, "select": "display_name"}
+        url = f"https://api.openalex.org/works?{urllib.parse.urlencode(params)}"
+        data = json.loads(fetch(url, timeout=timeout))
+        candidates = data.get("results", []) or []
+        if not candidates:
+            return {"status": "fail", "reason": "no matching title found on OpenAlex"}
+        sim, matched = _best_title_match(candidates, title, title_field="display_name")
+        status = "pass" if sim >= 0.9 else "fail"
+        return {"status": status, "similarity": round(sim, 3), "matched_title": matched}
+    except Exception as e:
+        return {"status": "skipped", "reason": str(e)}
+
+
+def _combine_title_checks(s2_result, openalex_result):
+    """Merge the two independent title cross-checks into one verdict-relevant
+    check. Coverage differs between S2 and OpenAlex, so one source not
+    indexing a (real) paper yet is common and must not, by itself, read as
+    "actively contradicted" the way a single-source design would have taken
+    it -- a pass from either source is enough; it only counts as an active
+    fail once neither source could corroborate it and at least one of them
+    genuinely searched (wasn't just skipped for network reasons)."""
+    sources = {"semantic_scholar": s2_result, "openalex": openalex_result}
+    passing = [r for r in sources.values() if r["status"] == "pass"]
+    if passing:
+        best = max(passing, key=lambda r: r.get("similarity", 0))
+        return {"status": "pass", "sources": sources, "similarity": best.get("similarity"),
+                "matched_title": best.get("matched_title")}
+    non_skipped = [r for r in sources.values() if r["status"] != "skipped"]
+    if non_skipped:
+        return {"status": "fail", "sources": sources,
+                "reason": "no title match on either Semantic Scholar or OpenAlex"}
+    return {"status": "skipped", "sources": sources, "reason": "both sources unavailable"}
+
+
 def verify(title=None, arxiv_id=None, doi=None):
     checks = {}
     if arxiv_id:
@@ -94,7 +135,10 @@ def verify(title=None, arxiv_id=None, doi=None):
     if doi:
         checks["doi_check"] = check_doi(doi)
     if title:
-        checks["title_cross_source_check"] = check_title_cross_source(title)
+        checks["title_cross_source_check"] = _combine_title_checks(
+            check_title_cross_source_s2(title),
+            check_title_cross_source_openalex(title),
+        )
 
     passed = [c for c in checks.values() if c["status"] == "pass"]
     failed = [c for c in checks.values() if c["status"] == "fail"]