zhoujie ea69788d0c 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>
2026-07-21 02:53:27 -10:00

173 lines
7.2 KiB
Python

#!/usr/bin/env python3
"""
Archive a finished literature-search-verify session into a permanent,
project-level folder instead of leaving results sitting in the skill's
own scratch output/ directory (which is easy to lose track of across
sessions and isn't meant to be a durable deliverable location).
Bundles the verified BibTeX file -- and, if given, any downloaded PDFs --
into <project-root>/references/<topic-slug>/, and writes a README.md
index (entry list, suspect/unverified entries flagged separately, free-
text coverage notes) so a future session or a human can find and trust
what's there without re-reading the conversation that produced it.
No third-party dependencies; uses only the standard library.
CLI usage (run from the project root; scratch input lives under
output/literature-search-verify/<topic-slug>/, matching the slug this
script derives from the topic argument):
python3 .claude/skills/literature-search-verify/scripts/archive_references.py \\
"UAV aeromagnetic compensation" \\
--bib output/literature-search-verify/uav_aeromagnetic_compensation/uav_aeromagnetic_compensation_final.bib \\
--project-root . \\
--pdfs-dir output/literature-search-verify/uav_aeromagnetic_compensation/pdfs \\
--suspect "Some fabricated-looking title|DOI resolves but venue is topically unrelated" \\
--notes "Kalman-filter and GA/PSO angles searched, no on-topic hits found."
Output: prints the path of the archive directory that was created/updated.
"""
import argparse
import os
import re
import shutil
import sys
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)
return text.strip("_")[:60] or "references"
def parse_bib_entries(bib_path):
"""Minimal BibTeX parser -- just enough to pull key/title/year/venue/doi/note
(plus the raw entry text, for reordering) for the README index. Not a
general-purpose BibTeX parser."""
with open(bib_path, encoding="utf-8") as f:
content = f.read()
entries = []
for m in re.finditer(r"@(\w+)\{([^,\n]+),(.*?)\n\}", content, re.S):
entry_type, key, body = m.groups()
fields = {}
for fm in re.finditer(r"(\w+)\s*=\s*\{(.*?)\}\s*,?\s*(?=\n\s*\w+\s*=|\n\Z|\Z)", body, re.S):
fields[fm.group(1).lower()] = re.sub(r"\s+", " ", fm.group(2)).strip()
entries.append({"type": entry_type, "key": key.strip(), "raw": m.group(0).strip(), **fields})
return entries
def year_sort_key(entry):
"""Chronological order, oldest first; entries with no parseable year sort last."""
year_str = re.sub(r"[^0-9]", "", entry.get("year", "") or "")
year = int(year_str) if year_str else 9999
return (year, entry.get("key", ""))
def build_readme(topic, entries, pdf_count, suspect, notes):
lines = []
lines.append(f"# {topic} — literature archive")
lines.append("")
lines.append(f"Archived: {date.today().isoformat()}")
lines.append(f"Verified entries: {len(entries)}")
lines.append(f"PDFs bundled: {pdf_count}")
lines.append("")
lines.append(
"Every entry in `references.bib` passed independent verification "
"(arXiv ID / DOI resolution and/or cross-source title match, "
"similarity >= 0.9) via the literature-search-verify skill before "
"being archived here. Citation keys follow the surname+year "
"convention and are stable -- the paper-writing-grounded skill's "
"`\\cite{}` calls should match these keys directly."
)
lines.append("")
lines.append("## Entries (chronological, oldest first)")
lines.append("")
for e in entries:
title = e.get("title", "?")
year = e.get("year", "?")
venue = e.get("journal") or e.get("booktitle") or e.get("school") or ""
doi = e.get("doi", "")
note = e.get("note", "")
line = f"- **{e['key']}** ({year}) — {title}"
if venue:
line += f". *{venue}*"
if doi:
line += f". DOI: {doi}"
lines.append(line)
if note:
lines.append(f" - Note: {note}")
if suspect:
lines.append("")
lines.append("## Flagged during search — NOT included above, do not cite")
lines.append("")
for s in suspect:
parts = s.split("|", 1)
title = parts[0].strip()
reason = parts[1].strip() if len(parts) > 1 else ""
lines.append(f"- {title}" + (f"{reason}" if reason else ""))
if notes:
lines.append("")
lines.append("## Search coverage notes")
lines.append("")
lines.append(notes)
return "\n".join(lines) + "\n"
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("topic", help="Human-readable topic name, e.g. \"UAV aeromagnetic compensation\"")
ap.add_argument("--bib", required=True, help="path to the curated/verified .bib file to archive")
ap.add_argument("--project-root", default=".", help="project root; archive is written under <root>/references/<slug>/")
ap.add_argument("--pdfs-dir", default=None, help="optional folder of open-access PDFs to copy alongside the bib")
ap.add_argument("--suspect", action="append", default=[], help="title|reason of a suspect/unverified entry to log; repeatable")
ap.add_argument("--notes", default=None, help="free-text notes on search coverage/gaps for the README")
args = ap.parse_args()
if not os.path.isfile(args.bib):
print(f"error: bib file not found: {args.bib}", file=sys.stderr)
sys.exit(1)
slug = slugify(args.topic)
archive_dir = os.path.join(args.project_root, "references", slug)
os.makedirs(archive_dir, exist_ok=True)
bib_dest = os.path.join(archive_dir, "references.bib")
shutil.copyfile(args.bib, bib_dest)
entries = parse_bib_entries(bib_dest)
entries.sort(key=year_sort_key)
# Rewrite the archived .bib in chronological order (oldest first) so the
# file itself, not just the README, reads as a timeline.
header = f"% {args.topic} -- verified references, chronological order\n% Archived {date.today().isoformat()}\n\n"
with open(bib_dest, "w", encoding="utf-8") as f:
f.write(header)
f.write("\n\n".join(e["raw"] for e in entries))
f.write("\n")
pdf_count = 0
if args.pdfs_dir and os.path.isdir(args.pdfs_dir):
pdf_dest_dir = os.path.join(archive_dir, "pdfs")
os.makedirs(pdf_dest_dir, exist_ok=True)
for fn in sorted(os.listdir(args.pdfs_dir)):
if fn.lower().endswith(".pdf"):
shutil.copyfile(os.path.join(args.pdfs_dir, fn), os.path.join(pdf_dest_dir, fn))
pdf_count += 1
readme = build_readme(args.topic, entries, pdf_count, args.suspect, args.notes)
with open(os.path.join(archive_dir, "README.md"), "w", encoding="utf-8") as f:
f.write(readme)
print(archive_dir)
if __name__ == "__main__":
main()