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
531 lines
19 KiB
Python
531 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
文献库管理器 - 教育AI研究项目
|
||
功能：
|
||
1. 添加/删除/更新文献条目
|
||
2. 搜索和筛选文献
|
||
3. 生成统计报告
|
||
4. 导出引用列表（APA/MLA/Chicago/GB/T 7714）
|
||
5. 维护索引和统计
|
||
6. DOI批量验证
|
||
|
||
用法示例：
|
||
python 文献库管理器.py --stats # 显示统计摘要
|
||
python 文献库管理器.py --search LLM # 搜索关键词
|
||
python 文献库管理器.py --list all # 列出所有文献
|
||
python 文献库管理器.py --export apa # 导出APA引用列表
|
||
python 文献库管理器.py --add-doi 10.xxx # 通过DOI添加文献
|
||
python 文献库管理器.py --verify # 验证所有DOI状态
|
||
|
||
作者：狗剩
|
||
创建时间：2026-04-04
|
||
更新时间：2026-04-05
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import argparse
|
||
from typing import Dict, List, Optional, Any
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
|
||
class LiteratureManager:
|
||
"""文献库管理器类"""
|
||
|
||
def __init__(self, db_path: str = None):
|
||
if db_path is None:
|
||
current_dir = Path(__file__).parent.parent
|
||
db_path = current_dir / "文献索引数据库.json"
|
||
|
||
self.db_path = Path(db_path)
|
||
self.data = self._load_database()
|
||
|
||
# ─────────────────────── 数据库读写 ───────────────────────
|
||
|
||
def _load_database(self) -> Dict:
|
||
if self.db_path.exists():
|
||
try:
|
||
with open(self.db_path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except json.JSONDecodeError:
|
||
print("⚠️ 数据库文件损坏，创建新数据库")
|
||
return self._create_empty_database()
|
||
else:
|
||
print(f"📝 数据库不存在，创建新数据库: {self.db_path}")
|
||
return self._create_empty_database()
|
||
|
||
def _create_empty_database(self) -> Dict:
|
||
return {
|
||
"metadata": {
|
||
"version": "1.1",
|
||
"created": datetime.now().strftime("%Y-%m-%d"),
|
||
"updated": datetime.now().strftime("%Y-%m-%d"),
|
||
"description": "教育AI研究项目文献索引数据库",
|
||
"total_entries": 0,
|
||
"last_entry_id": 0
|
||
},
|
||
"schema": {},
|
||
"entries": [],
|
||
"indexes": {
|
||
"by_year": {},
|
||
"by_journal": {},
|
||
"by_technology": {},
|
||
"by_education_level": {},
|
||
"by_reliability": {}
|
||
},
|
||
"statistics": {
|
||
"total_by_year": {},
|
||
"total_by_journal": {},
|
||
"total_by_technology": {},
|
||
"high_quality_entries": 0
|
||
}
|
||
}
|
||
|
||
def _save_database(self):
|
||
self.data["metadata"]["updated"] = datetime.now().strftime("%Y-%m-%d")
|
||
self.data["metadata"]["total_entries"] = len(self.data["entries"])
|
||
with open(self.db_path, 'w', encoding='utf-8') as f:
|
||
json.dump(self.data, f, ensure_ascii=False, indent=2)
|
||
|
||
# ─────────────────────── 添加文献 ───────────────────────
|
||
|
||
def add_entry(self, entry: Dict) -> int:
|
||
"""添加文献条目，返回新条目ID"""
|
||
last_id = self.data["metadata"].get("last_entry_id", 0)
|
||
new_id = last_id + 1
|
||
entry["entry_id"] = new_id
|
||
entry.setdefault("added_date", datetime.now().strftime("%Y-%m-%d"))
|
||
entry.setdefault("added_by", "狗剩")
|
||
self.data["entries"].append(entry)
|
||
self.data["metadata"]["last_entry_id"] = new_id
|
||
self._rebuild_indexes()
|
||
self._save_database()
|
||
print(f"✅ 文献已添加，ID: {new_id} — {entry.get('title', '无标题')[:50]}")
|
||
return new_id
|
||
|
||
def add_from_doi(self, doi: str) -> Optional[int]:
|
||
"""通过DOI自动获取元数据并添加"""
|
||
try:
|
||
from doi_resolver import DOIResolver
|
||
except ImportError:
|
||
print("⚠️ 无法导入DOIResolver，请手动补充元数据")
|
||
entry = {
|
||
"doi": doi,
|
||
"title": f"[待补充] {doi}",
|
||
"authors": [],
|
||
"journal": "",
|
||
"year": None,
|
||
"keywords": [],
|
||
"url": f"https://doi.org/{doi}",
|
||
"research_type": "",
|
||
"education_level": "",
|
||
"ai_technology": "",
|
||
"quality_score": 0,
|
||
"reliability": "待验证",
|
||
"notes": "通过DOI添加，元数据待补充",
|
||
"tags": [],
|
||
"related_entries": []
|
||
}
|
||
return self.add_entry(entry)
|
||
|
||
resolver = DOIResolver()
|
||
metadata = resolver.resolve_doi(doi)
|
||
|
||
if not metadata:
|
||
print(f"❌ DOI解析失败: {doi}")
|
||
return None
|
||
|
||
entry = {
|
||
"doi": metadata.get("doi", doi),
|
||
"title": metadata.get("title", ""),
|
||
"authors": metadata.get("authors", []),
|
||
"journal": metadata.get("journal", ""),
|
||
"year": metadata.get("year"),
|
||
"volume": metadata.get("volume", ""),
|
||
"issue": metadata.get("issue", ""),
|
||
"pages": metadata.get("pages", ""),
|
||
"abstract": metadata.get("abstract", ""),
|
||
"keywords": [],
|
||
"url": metadata.get("url", f"https://doi.org/{doi}"),
|
||
"pdf_path": "",
|
||
"citation_count": metadata.get("citation_count", 0),
|
||
"research_type": "",
|
||
"education_level": "",
|
||
"ai_technology": "",
|
||
"quality_score": 80,
|
||
"reliability": "高",
|
||
"notes": f"通过DOI自动导入，来源: {metadata.get('source', 'API')}",
|
||
"tags": [],
|
||
"related_entries": []
|
||
}
|
||
|
||
return self.add_entry(entry)
|
||
|
||
# ─────────────────────── 搜索 ───────────────────────
|
||
|
||
def search(self, query: str, fields: List[str] = None) -> List[Dict]:
|
||
"""全文搜索"""
|
||
if fields is None:
|
||
fields = ["title", "abstract", "keywords", "tags", "journal", "authors", "notes"]
|
||
|
||
query = query.lower()
|
||
results = []
|
||
|
||
for entry in self.data["entries"]:
|
||
matched = False
|
||
for field in fields:
|
||
val = entry.get(field, "")
|
||
if isinstance(val, list):
|
||
val = " ".join(str(v) for v in val)
|
||
if query in str(val).lower():
|
||
matched = True
|
||
break
|
||
if matched:
|
||
results.append(entry)
|
||
|
||
return results
|
||
|
||
def filter_by(self, **kwargs) -> List[Dict]:
|
||
"""按字段过滤，支持 year/journal/technology/reliability/education_level"""
|
||
results = self.data["entries"]
|
||
for key, value in kwargs.items():
|
||
results = [e for e in results if str(e.get(key, "")).lower() == str(value).lower()]
|
||
return results
|
||
|
||
def get_entry(self, entry_id: int) -> Optional[Dict]:
|
||
for e in self.data["entries"]:
|
||
if e.get("entry_id") == entry_id:
|
||
return e
|
||
return None
|
||
|
||
def get_unverified(self) -> List[Dict]:
|
||
"""获取所有待验证的条目"""
|
||
return [e for e in self.data["entries"] if e.get("reliability") == "待验证"]
|
||
|
||
# ─────────────────────── 引用格式 ───────────────────────
|
||
|
||
def format_citation(self, entry: Dict, style: str = "apa") -> str:
|
||
"""生成引用格式字符串"""
|
||
authors = entry.get("authors", [])
|
||
title = entry.get("title", "Unknown Title")
|
||
journal = entry.get("journal", "")
|
||
year = entry.get("year", "n.d.")
|
||
volume = entry.get("volume", "")
|
||
issue = entry.get("issue", "")
|
||
pages = entry.get("pages", "")
|
||
doi = entry.get("doi", "")
|
||
|
||
style = style.lower()
|
||
|
||
if style == "apa":
|
||
if len(authors) == 0:
|
||
a = "[Author unknown]"
|
||
elif len(authors) == 1:
|
||
a = authors[0].split(",")[0]
|
||
elif len(authors) == 2:
|
||
a = f"{authors[0].split(',')[0]} & {authors[1].split(',')[0]}"
|
||
else:
|
||
a = f"{authors[0].split(',')[0]} et al."
|
||
c = f"{a} ({year}). {title}."
|
||
if journal:
|
||
c += f" *{journal}*"
|
||
if volume:
|
||
c += f", *{volume}*"
|
||
if issue:
|
||
c += f"({issue})"
|
||
if pages:
|
||
c += f", {pages}"
|
||
c += "."
|
||
if doi and "XXX" not in doi:
|
||
c += f" https://doi.org/{doi}"
|
||
return c
|
||
|
||
elif style == "mla":
|
||
a = authors[0] if authors else "Unknown"
|
||
c = f'{a}. "{title}." *{journal}*'
|
||
if volume:
|
||
c += f", vol. {volume}"
|
||
if issue:
|
||
c += f", no. {issue}"
|
||
c += f", {year}"
|
||
if pages:
|
||
c += f", pp. {pages}"
|
||
c += "."
|
||
if doi and "XXX" not in doi:
|
||
c += f" doi:{doi}."
|
||
return c
|
||
|
||
elif style == "chicago":
|
||
a = authors[0] if authors else "Unknown"
|
||
c = f'{a}. "{title}." *{journal}*'
|
||
if volume:
|
||
c += f" {volume}"
|
||
if issue:
|
||
c += f", no. {issue}"
|
||
c += f" ({year})"
|
||
if pages:
|
||
c += f": {pages}"
|
||
c += "."
|
||
if doi and "XXX" not in doi:
|
||
c += f" https://doi.org/{doi}."
|
||
return c
|
||
|
||
elif style in ("gb", "gb/t 7714", "gbt"):
|
||
if len(authors) == 0:
|
||
a = "佚名"
|
||
elif len(authors) > 3:
|
||
a = f"{authors[0]} 等"
|
||
else:
|
||
a = ", ".join(authors)
|
||
c = f"{a}. {title}[J]. {journal}"
|
||
if year:
|
||
c += f", {year}"
|
||
if volume:
|
||
c += f", {volume}"
|
||
if issue:
|
||
c += f"({issue})"
|
||
if pages:
|
||
c += f": {pages}"
|
||
c += "."
|
||
if doi and "XXX" not in doi:
|
||
c += f" DOI: {doi}."
|
||
return c
|
||
|
||
return f"[{entry.get('entry_id')}] {title} ({year}). {journal}."
|
||
|
||
# ─────────────────────── 导出 ───────────────────────
|
||
|
||
def export_citations(self, style: str = "apa", filter_verified: bool = False,
|
||
output_path: str = None) -> str:
|
||
"""导出所有文献的引用列表"""
|
||
entries = self.data["entries"]
|
||
if filter_verified:
|
||
entries = [e for e in entries if e.get("reliability") == "高"]
|
||
|
||
lines = [f"# 教育AI研究 — 参考文献列表（{style.upper()}格式）",
|
||
f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}",
|
||
f"总计：{len(entries)} 条（{'仅高可靠' if filter_verified else '全部'}）",
|
||
""]
|
||
|
||
for entry in sorted(entries, key=lambda e: (e.get("year") or 0), reverse=True):
|
||
citation = self.format_citation(entry, style)
|
||
note = ""
|
||
if entry.get("reliability") == "待验证":
|
||
note = " ⚠️[待验证]"
|
||
lines.append(f"- {citation}{note}")
|
||
|
||
text = "\n".join(lines)
|
||
|
||
if output_path:
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
f.write(text)
|
||
print(f"✅ 引用列表已导出: {output_path}")
|
||
else:
|
||
print(text)
|
||
|
||
return text
|
||
|
||
# ─────────────────────── 统计 ───────────────────────
|
||
|
||
def print_stats(self):
|
||
"""打印统计摘要"""
|
||
entries = self.data["entries"]
|
||
total = len(entries)
|
||
verified = sum(1 for e in entries if e.get("reliability") == "高")
|
||
need_verify = sum(1 for e in entries if e.get("reliability") == "待验证")
|
||
with_doi = sum(1 for e in entries if e.get("doi") and "XXX" not in e.get("doi", ""))
|
||
high_q = sum(1 for e in entries if (e.get("quality_score") or 0) >= 85)
|
||
|
||
print("\n" + "=" * 55)
|
||
print(" 📚 教育AI研究文献库 — 统计摘要")
|
||
print("=" * 55)
|
||
print(f" 总文献数： {total}")
|
||
print(f" 高可靠来源： {verified} ({verified*100//total if total else 0}%)")
|
||
print(f" 待验证： {need_verify}")
|
||
print(f" 有效DOI： {with_doi}")
|
||
print(f" 高质量(≥85)： {high_q}")
|
||
print()
|
||
|
||
# 按技术分类
|
||
tech_counts = {}
|
||
for e in entries:
|
||
t = e.get("ai_technology", "其他")
|
||
if not t:
|
||
t = "其他"
|
||
for tag in t.split("/"):
|
||
tag = tag.strip()
|
||
tech_counts[tag] = tech_counts.get(tag, 0) + 1
|
||
|
||
print(" 技术分类：")
|
||
for t, c in sorted(tech_counts.items(), key=lambda x: -x[1]):
|
||
print(f" {t:<20} {c} 篇")
|
||
|
||
print()
|
||
|
||
# 按年份
|
||
year_counts = {}
|
||
for e in entries:
|
||
y = str(e.get("year", "未知"))
|
||
year_counts[y] = year_counts.get(y, 0) + 1
|
||
print(" 年份分布：")
|
||
for y, c in sorted(year_counts.items(), reverse=True):
|
||
bar = "█" * c
|
||
print(f" {y} {bar} ({c})")
|
||
|
||
print("=" * 55 + "\n")
|
||
|
||
# ─────────────────────── 列表 ───────────────────────
|
||
|
||
def print_list(self, entries: List[Dict] = None, verbose: bool = False):
|
||
"""打印文献列表"""
|
||
if entries is None:
|
||
entries = self.data["entries"]
|
||
|
||
if not entries:
|
||
print(" （无结果）")
|
||
return
|
||
|
||
for e in entries:
|
||
flag = "✅" if e.get("reliability") == "高" else "⚠️"
|
||
doi_note = ""
|
||
if e.get("doi") and "XXX" not in e.get("doi", ""):
|
||
doi_note = f" DOI: {e['doi']}"
|
||
print(f"\n [{e.get('entry_id')}] {flag} {e.get('title', '无标题')[:60]}")
|
||
print(f" {e.get('journal', '')} ({e.get('year', '?')}){doi_note}")
|
||
if verbose:
|
||
if e.get("tags"):
|
||
print(f" 标签: {', '.join(e['tags'][:5])}")
|
||
if e.get("notes"):
|
||
print(f" 备注: {e['notes'][:80]}")
|
||
|
||
# ─────────────────────── 验证 ───────────────────────
|
||
|
||
def verify_dois(self):
|
||
"""列出所有DOI状态"""
|
||
print("\n📋 DOI 验证状态：\n")
|
||
for e in self.data["entries"]:
|
||
doi = e.get("doi", "")
|
||
title = e.get("title", "")[:45]
|
||
if not doi:
|
||
status = "❌ 无DOI"
|
||
elif "XXX" in doi:
|
||
status = "⚠️ DOI不完整（含XXX占位）"
|
||
elif re.match(r'^10\.\d{4,}/.+$', doi):
|
||
status = "✅ DOI格式有效"
|
||
else:
|
||
status = "❓ 格式异常"
|
||
print(f" [{e.get('entry_id'):>2}] {status:<28} {title}")
|
||
|
||
# ─────────────────────── 索引 ───────────────────────
|
||
|
||
def _rebuild_indexes(self):
|
||
indexes = {
|
||
"by_year": {},
|
||
"by_journal": {},
|
||
"by_technology": {},
|
||
"by_education_level": {},
|
||
"by_reliability": {}
|
||
}
|
||
|
||
for e in self.data["entries"]:
|
||
eid = e.get("entry_id")
|
||
|
||
y = str(e.get("year", "未知"))
|
||
indexes["by_year"].setdefault(y, []).append(eid)
|
||
|
||
j = e.get("journal", "未知")
|
||
indexes["by_journal"].setdefault(j, []).append(eid)
|
||
|
||
for t in (e.get("ai_technology") or "其他").split("/"):
|
||
t = t.strip()
|
||
indexes["by_technology"].setdefault(t, []).append(eid)
|
||
|
||
lv = e.get("education_level", "未知")
|
||
indexes["by_education_level"].setdefault(lv, []).append(eid)
|
||
|
||
rel = e.get("reliability", "待验证")
|
||
indexes["by_reliability"].setdefault(rel, []).append(eid)
|
||
|
||
self.data["indexes"] = indexes
|
||
|
||
|
||
# ─────────────────────── CLI ───────────────────────
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="教育AI研究文献库管理器",
|
||
formatter_class=argparse.RawTextHelpFormatter
|
||
)
|
||
parser.add_argument("--stats", action="store_true", help="显示统计摘要")
|
||
parser.add_argument("--list", metavar="FILTER",
|
||
help="列出文献（all / verified / unverified / <年份>）")
|
||
parser.add_argument("--verbose", "-v", action="store_true", help="显示详细信息")
|
||
parser.add_argument("--search", metavar="QUERY", help="搜索文献")
|
||
parser.add_argument("--export", metavar="STYLE",
|
||
help="导出引用列表（apa / mla / chicago / gb）")
|
||
parser.add_argument("--export-verified", action="store_true", help="仅导出高可靠文献")
|
||
parser.add_argument("--output", metavar="FILE", help="导出到文件")
|
||
parser.add_argument("--add-doi", metavar="DOI", help="通过DOI添加文献")
|
||
parser.add_argument("--verify", action="store_true", help="验证所有DOI状态")
|
||
parser.add_argument("--show", metavar="ID", type=int, help="显示指定ID的文献详情")
|
||
parser.add_argument("--db", metavar="PATH", help="指定数据库路径")
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 初始化管理器
|
||
db_path = args.db if args.db else None
|
||
mgr = LiteratureManager(db_path)
|
||
|
||
if args.stats:
|
||
mgr.print_stats()
|
||
|
||
elif args.list:
|
||
if args.list == "all":
|
||
entries = mgr.data["entries"]
|
||
elif args.list == "verified":
|
||
entries = [e for e in mgr.data["entries"] if e.get("reliability") == "高"]
|
||
elif args.list == "unverified":
|
||
entries = mgr.get_unverified()
|
||
else:
|
||
entries = mgr.filter_by(year=args.list)
|
||
print(f"\n共 {len(entries)} 条：")
|
||
mgr.print_list(entries, verbose=args.verbose)
|
||
|
||
elif args.search:
|
||
results = mgr.search(args.search)
|
||
print(f"\n搜索 '{args.search}' 找到 {len(results)} 条：")
|
||
mgr.print_list(results, verbose=args.verbose)
|
||
|
||
elif args.export:
|
||
mgr.export_citations(
|
||
style=args.export,
|
||
filter_verified=args.export_verified,
|
||
output_path=args.output
|
||
)
|
||
|
||
elif args.add_doi:
|
||
mgr.add_from_doi(args.add_doi)
|
||
|
||
elif args.verify:
|
||
mgr.verify_dois()
|
||
|
||
elif args.show:
|
||
entry = mgr.get_entry(args.show)
|
||
if entry:
|
||
print(json.dumps(entry, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(f"❌ 未找到ID: {args.show}")
|
||
|
||
else:
|
||
# 默认：显示统计
|
||
mgr.print_stats()
|
||
print("提示：使用 --help 查看所有命令")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|