hehaiguang1123
a6f05ab2d5
- Phase 0: AGENTS.md cleanup (dedup quotes, renumber sections, merge qmd) - Phase 1: typed relations (manage-relations.py, graph-search.py, check-staleness.py, detect-conflicts.py) - Phase 2: frontmatter validator, weekly lint, knowledge promotion, git hooks - Fix .gitignore to track tools/ and .githooks/ - Fix git remote URL (remove plaintext token) - New wiki pages: 504 pages, 34 raw sources
433 lines
14 KiB
Python
433 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
DOI解析工具 - 教育AI研究项目文献库
|
|
功能：
|
|
1. 解析DOI获取文献元数据
|
|
2. 生成标准引用格式
|
|
3. 验证DOI有效性
|
|
4. 批量处理DOI列表
|
|
|
|
作者：狗剩
|
|
创建时间：2026-04-04
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import urllib.request
|
|
import urllib.error
|
|
from typing import Dict, List, Optional, Tuple
|
|
from datetime import datetime
|
|
|
|
|
|
class DOIResolver:
|
|
"""DOI解析器类"""
|
|
|
|
def __init__(self):
|
|
self.base_url = "https://doi.org/"
|
|
self.crossref_api = "https://api.crossref.org/works/"
|
|
self.datacite_api = "https://api.datacite.org/dois/"
|
|
|
|
def validate_doi(self, doi: str) -> bool:
|
|
"""
|
|
验证DOI格式是否有效
|
|
|
|
Args:
|
|
doi: DOI字符串
|
|
|
|
Returns:
|
|
bool: 是否有效
|
|
"""
|
|
if not doi:
|
|
return False
|
|
|
|
# 清理DOI字符串
|
|
doi = doi.strip()
|
|
|
|
# 移除可能的URL前缀
|
|
if doi.startswith('http'):
|
|
doi = doi.split('doi.org/')[-1]
|
|
|
|
# DOI正则表达式模式
|
|
pattern = r'^10\.\d{4,}\/.+$'
|
|
return bool(re.match(pattern, doi))
|
|
|
|
def clean_doi(self, doi: str) -> str:
|
|
"""
|
|
清理DOI字符串，提取纯DOI
|
|
|
|
Args:
|
|
doi: 可能包含URL的DOI字符串
|
|
|
|
Returns:
|
|
str: 纯DOI
|
|
"""
|
|
doi = doi.strip()
|
|
|
|
# 移除URL前缀
|
|
if 'doi.org/' in doi:
|
|
doi = doi.split('doi.org/')[-1]
|
|
elif doi.startswith('http'):
|
|
doi = doi.split('/')[-1]
|
|
|
|
return doi
|
|
|
|
def resolve_doi(self, doi: str, timeout: int = 10) -> Optional[Dict]:
|
|
"""
|
|
解析DOI获取文献元数据
|
|
|
|
Args:
|
|
doi: DOI字符串
|
|
timeout: 请求超时时间（秒）
|
|
|
|
Returns:
|
|
Dict: 文献元数据，失败返回None
|
|
"""
|
|
doi = self.clean_doi(doi)
|
|
|
|
if not self.validate_doi(doi):
|
|
print(f"❌ 无效的DOI格式: {doi}")
|
|
return None
|
|
|
|
try:
|
|
# 尝试Crossref API
|
|
metadata = self._fetch_from_crossref(doi, timeout)
|
|
if metadata:
|
|
return metadata
|
|
|
|
# 尝试DataCite API
|
|
metadata = self._fetch_from_datacite(doi, timeout)
|
|
if metadata:
|
|
return metadata
|
|
|
|
print(f"⚠️ 无法解析DOI: {doi}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"❌ 解析DOI时出错: {doi}, 错误: {str(e)}")
|
|
return None
|
|
|
|
def _fetch_from_crossref(self, doi: str, timeout: int) -> Optional[Dict]:
|
|
"""从Crossref API获取元数据"""
|
|
try:
|
|
url = f"{self.crossref_api}{doi}"
|
|
headers = {
|
|
'User-Agent': '教育AI研究项目文献库 (mailto:research@example.com)'
|
|
}
|
|
|
|
request = urllib.request.Request(url, headers=headers)
|
|
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
data = json.loads(response.read().decode('utf-8'))
|
|
|
|
if data.get('status') == 'ok':
|
|
message = data.get('message', {})
|
|
return self._parse_crossref_data(message)
|
|
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
return None
|
|
raise
|
|
|
|
return None
|
|
|
|
def _fetch_from_datacite(self, doi: str, timeout: int) -> Optional[Dict]:
|
|
"""从DataCite API获取元数据"""
|
|
try:
|
|
url = f"{self.datacite_api}{doi}"
|
|
headers = {
|
|
'User-Agent': '教育AI研究项目文献库 (mailto:research@example.com)'
|
|
}
|
|
|
|
request = urllib.request.Request(url, headers=headers)
|
|
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
data = json.loads(response.read().decode('utf-8'))
|
|
|
|
if 'data' in data:
|
|
return self._parse_datacite_data(data['data'])
|
|
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
return None
|
|
raise
|
|
|
|
return None
|
|
|
|
def _parse_crossref_data(self, message: Dict) -> Dict:
|
|
"""解析Crossref返回的数据"""
|
|
# 提取作者信息
|
|
authors = []
|
|
for author in message.get('author', []):
|
|
given = author.get('given', '')
|
|
family = author.get('family', '')
|
|
if given and family:
|
|
authors.append(f"{family}, {given}")
|
|
elif family:
|
|
authors.append(family)
|
|
|
|
# 提取日期
|
|
published = message.get('published-print', message.get('published-online', {}))
|
|
year = published.get('date-parts', [[None]])[0][0] if published else None
|
|
|
|
# 构建元数据
|
|
metadata = {
|
|
'doi': message.get('DOI', ''),
|
|
'title': message.get('title', [''])[0] if message.get('title') else '',
|
|
'authors': authors,
|
|
'journal': message.get('container-title', [''])[0] if message.get('container-title') else '',
|
|
'year': year,
|
|
'volume': message.get('volume', ''),
|
|
'issue': message.get('issue', ''),
|
|
'pages': message.get('page', ''),
|
|
'abstract': message.get('abstract', ''),
|
|
'publisher': message.get('publisher', ''),
|
|
'url': f"https://doi.org/{message.get('DOI', '')}",
|
|
'citation_count': message.get('is-referenced-by-count', 0),
|
|
'source': 'Crossref',
|
|
'parsed_date': datetime.now().isoformat()
|
|
}
|
|
|
|
return metadata
|
|
|
|
def _parse_datacite_data(self, data: Dict) -> Dict:
|
|
"""解析DataCite返回的数据"""
|
|
attributes = data.get('attributes', {})
|
|
|
|
# 提取作者信息
|
|
authors = []
|
|
for creator in attributes.get('creators', []):
|
|
name = creator.get('name', '')
|
|
if name:
|
|
authors.append(name)
|
|
|
|
# 提取日期
|
|
dates = attributes.get('dates', [])
|
|
year = None
|
|
for date in dates:
|
|
if date.get('dateType') == 'Issued':
|
|
year = date.get('date', '')[:4]
|
|
break
|
|
|
|
# 构建元数据
|
|
metadata = {
|
|
'doi': attributes.get('doi', ''),
|
|
'title': attributes.get('titles', [{}])[0].get('title', ''),
|
|
'authors': authors,
|
|
'journal': attributes.get('container', {}).get('title', ''),
|
|
'year': int(year) if year and year.isdigit() else None,
|
|
'publisher': attributes.get('publisher', ''),
|
|
'url': f"https://doi.org/{attributes.get('doi', '')}",
|
|
'source': 'DataCite',
|
|
'parsed_date': datetime.now().isoformat()
|
|
}
|
|
|
|
return metadata
|
|
|
|
def generate_citation(self, metadata: Dict, style: str = 'apa') -> str:
|
|
"""
|
|
生成标准引用格式
|
|
|
|
Args:
|
|
metadata: 文献元数据
|
|
style: 引用格式 (apa, mla, chicago, gb/t 7714)
|
|
|
|
Returns:
|
|
str: 格式化引用
|
|
"""
|
|
authors = metadata.get('authors', [])
|
|
title = metadata.get('title', '')
|
|
journal = metadata.get('journal', '')
|
|
year = metadata.get('year', '')
|
|
volume = metadata.get('volume', '')
|
|
issue = metadata.get('issue', '')
|
|
pages = metadata.get('pages', '')
|
|
doi = metadata.get('doi', '')
|
|
|
|
if style.lower() == 'apa':
|
|
return self._format_apa(authors, title, journal, year, volume, issue, pages, doi)
|
|
elif style.lower() == 'mla':
|
|
return self._format_mla(authors, title, journal, year, volume, issue, pages, doi)
|
|
elif style.lower() == 'chicago':
|
|
return self._format_chicago(authors, title, journal, year, volume, issue, pages, doi)
|
|
elif style.lower() == 'gb' or style.lower() == 'gb/t 7714':
|
|
return self._format_gb(authors, title, journal, year, volume, issue, pages, doi)
|
|
else:
|
|
return self._format_apa(authors, title, journal, year, volume, issue, pages, doi)
|
|
|
|
def _format_apa(self, authors, title, journal, year, volume, issue, pages, doi):
|
|
"""APA格式"""
|
|
if len(authors) == 1:
|
|
author_str = authors[0].split(',')[0]
|
|
elif len(authors) == 2:
|
|
author_str = f"{authors[0].split(',')[0]} & {authors[1].split(',')[0]}"
|
|
elif len(authors) > 2:
|
|
author_str = f"{authors[0].split(',')[0]} et al."
|
|
else:
|
|
author_str = "Unknown"
|
|
|
|
citation = f"{author_str} ({year}). {title}."
|
|
if journal:
|
|
citation += f" *{journal}*"
|
|
if volume:
|
|
citation += f", *{volume}*"
|
|
if issue:
|
|
citation += f"({issue})"
|
|
if pages:
|
|
citation += f", {pages}"
|
|
citation += "."
|
|
if doi:
|
|
citation += f" https://doi.org/{doi}"
|
|
|
|
return citation
|
|
|
|
def _format_mla(self, authors, title, journal, year, volume, issue, pages, doi):
|
|
"""MLA格式"""
|
|
if authors:
|
|
author_str = authors[0]
|
|
else:
|
|
author_str = "Unknown"
|
|
|
|
citation = f'"{title}." *{journal}*, vol. {volume}, no. {issue}, {year}, pp. {pages}.'
|
|
if doi:
|
|
citation += f" doi:{doi}."
|
|
|
|
return citation
|
|
|
|
def _format_chicago(self, authors, title, journal, year, volume, issue, pages, doi):
|
|
"""Chicago格式"""
|
|
if authors:
|
|
author_str = authors[0]
|
|
else:
|
|
author_str = "Unknown"
|
|
|
|
citation = f'{author_str}. "{title}." *{journal}* {volume}, no. {issue} ({year}): {pages}.'
|
|
if doi:
|
|
citation += f" doi:{doi}."
|
|
|
|
return citation
|
|
|
|
def _format_gb(self, authors, title, journal, year, volume, issue, pages, doi):
|
|
"""GB/T 7714格式（中国国家标准）"""
|
|
if authors:
|
|
if len(authors) > 3:
|
|
author_str = f"{authors[0]} 等"
|
|
else:
|
|
author_str = ", ".join(authors)
|
|
else:
|
|
author_str = "佚名"
|
|
|
|
citation = f"{author_str}. {title}[J]. {journal}"
|
|
if year:
|
|
citation += f", {year}"
|
|
if volume:
|
|
citation += f", {volume}"
|
|
if issue:
|
|
citation += f"({issue})"
|
|
if pages:
|
|
citation += f": {pages}"
|
|
citation += "."
|
|
if doi:
|
|
citation += f" DOI: {doi}."
|
|
|
|
return citation
|
|
|
|
def batch_resolve(self, doi_list: List[str], delay: float = 1.0) -> List[Tuple[str, Optional[Dict]]]:
|
|
"""
|
|
批量解析DOI列表
|
|
|
|
Args:
|
|
doi_list: DOI列表
|
|
delay: 请求间隔（秒）
|
|
|
|
Returns:
|
|
List[Tuple]: (doi, metadata) 元组列表
|
|
"""
|
|
import time
|
|
|
|
results = []
|
|
total = len(doi_list)
|
|
|
|
print(f"开始批量解析 {total} 个DOI...")
|
|
|
|
for i, doi in enumerate(doi_list, 1):
|
|
print(f"[{i}/{total}] 解析: {doi}")
|
|
metadata = self.resolve_doi(doi)
|
|
results.append((doi, metadata))
|
|
|
|
if i < total:
|
|
time.sleep(delay)
|
|
|
|
success = sum(1 for _, m in results if m is not None)
|
|
print(f"\n解析完成: {success}/{total} 成功")
|
|
|
|
return results
|
|
|
|
|
|
def main():
|
|
"""主函数 - 命令行接口"""
|
|
import sys
|
|
|
|
resolver = DOIResolver()
|
|
|
|
if len(sys.argv) < 2:
|
|
print("用法:")
|
|
print(" python doi_resolver.py <DOI> [style]")
|
|
print(" python doi_resolver.py --batch <doi_list_file>")
|
|
print("")
|
|
print("示例:")
|
|
print(' python doi_resolver.py "10.1038/s41599-026-07019-z" apa')
|
|
print(' python doi_resolver.py --batch doi_list.txt')
|
|
sys.exit(1)
|
|
|
|
if sys.argv[1] == '--batch':
|
|
# 批量处理模式
|
|
if len(sys.argv) < 3:
|
|
print("错误: 请提供DOI列表文件路径")
|
|
sys.exit(1)
|
|
|
|
file_path = sys.argv[2]
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
doi_list = [line.strip() for line in f if line.strip()]
|
|
|
|
results = resolver.batch_resolve(doi_list)
|
|
|
|
# 保存结果
|
|
output_file = file_path.replace('.txt', '_resolved.json')
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
json.dump([{'doi': d, 'metadata': m} for d, m in results], f,
|
|
ensure_ascii=False, indent=2)
|
|
print(f"\n结果已保存: {output_file}")
|
|
|
|
except FileNotFoundError:
|
|
print(f"错误: 文件不存在: {file_path}")
|
|
sys.exit(1)
|
|
else:
|
|
# 单DOI处理模式
|
|
doi = sys.argv[1]
|
|
style = sys.argv[2] if len(sys.argv) > 2 else 'apa'
|
|
|
|
print(f"解析DOI: {doi}")
|
|
print("-" * 50)
|
|
|
|
metadata = resolver.resolve_doi(doi)
|
|
|
|
if metadata:
|
|
print("✅ 解析成功!")
|
|
print("\n元数据:")
|
|
for key, value in metadata.items():
|
|
if value:
|
|
print(f" {key}: {value}")
|
|
|
|
print(f"\n引用格式 ({style.upper()}):")
|
|
citation = resolver.generate_citation(metadata, style)
|
|
print(citation)
|
|
else:
|
|
print("❌ 解析失败")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|