zhoujie
ea69788d0c
新增 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>
93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
#!/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)
|