Files
llm_wiki/raw/教育AI研究/tools/verify_auto_sync.py
T
hehaiguang1123 a6f05ab2d5 Phase 0-2: Schema cleanup, typed relations, event-driven automation
- 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
2026-07-01 08:05:43 +08:00

387 lines
13 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import sys
import json
import datetime
from pathlib import Path
from typing import Dict, List, Tuple
# Set UTF-8 encoding for Windows console
if sys.platform == "win32":
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")
class KnowledgeSyncVerifier:
def __init__(self, project_root: Path):
self.project_root = project_root
self.edu_ai_path = project_root / "raw" / "教育AI研究"
self.issues = []
self.warnings = []
self.successes = []
def verify_directory_structure(self) -> bool:
print("📁 验证目录结构...")
required_dirs = [
"机构档案",
"深度研究报告",
"每周报告",
"知识卡片",
"文献库",
"案例分析",
"归档",
"outputs",
]
all_valid = True
for dir_name in required_dirs:
dir_path = self.edu_ai_path / dir_name
if not dir_path.exists():
self.issues.append(f"缺少必要目录: {dir_name}")
all_valid = False
print(f" ❌ 缺少: {dir_name}")
else:
self.successes.append(f"目录存在: {dir_name}")
print(f" ✅ 存在: {dir_name}")
return all_valid
def verify_knowledge_card_consistency(self) -> Tuple[int, int]:
print("\n📚 检查知识卡片一致性...")
cards_path = self.edu_ai_path / "知识卡片"
if not cards_path.exists():
self.issues.append("知识卡片目录不存在")
return 0, 0
card_files = list(cards_path.glob("*.md"))
print(f" 找到 {len(card_files)} 张知识卡片")
consistency_checks = {
"has_frontmatter": 0,
"has_categories": 0,
"has_tags": 0,
"has_type": 0,
}
for card_file in card_files:
try:
content = card_file.read_text(encoding="utf-8")
has_frontmatter = "---" in content[:200]
consistency_checks["has_frontmatter"] += 1 if has_frontmatter else 0
if has_frontmatter:
if "categories:" in content:
consistency_checks["has_categories"] += 1
if "tags:" in content:
consistency_checks["has_tags"] += 1
if "type:" in content:
consistency_checks["has_type"] += 1
except Exception as e:
self.warnings.append(f"无法读取卡片 {card_file.name}: {str(e)}")
total_checks = sum(consistency_checks.values())
max_checks = len(card_files) * len(consistency_checks)
consistency_rate = (total_checks / max_checks * 100) if max_checks > 0 else 0
print(
f" 前置元数据检查: {consistency_checks['has_frontmatter']}/{len(card_files)}"
)
print(
f" 类别字段检查: {consistency_checks['has_categories']}/{len(card_files)}"
)
print(f" 标签字段检查: {consistency_checks['has_tags']}/{len(card_files)}")
print(f" 类型字段检查: {consistency_checks['has_type']}/{len(card_files)}")
print(f" 📊 一致性得分: {consistency_rate:.1f}%")
if consistency_rate < 80:
self.issues.append(f"知识卡片一致性低于80%: {consistency_rate:.1f}%")
else:
self.successes.append(f"知识卡片一致性良好: {consistency_rate:.1f}%")
return total_checks, max_checks
def verify_institution_archives(self) -> Tuple[int, List[str]]:
print("\n🏢 检查机构档案信息...")
institutions_path = self.edu_ai_path / "机构档案"
if not institutions_path.exists():
self.issues.append("机构档案目录不存在")
return 0, []
inst_files = list(institutions_path.glob("*.md"))
print(f" 找到 {len(inst_files)} 个机构档案")
quality_scores = []
for inst_file in inst_files:
try:
content = inst_file.read_text(encoding="utf-8")
score = 0
total_checks = 10
if "---" in content[:200]:
score += 1
if "官网" in content or "官方网站" in content:
score += 1
if "研究重点" in content or "研究方向" in content:
score += 1
if "主要成果" in content or "研究成果" in content:
score += 1
if "联系方式" in content or "联系信息" in content:
score += 1
if len(content) > 1000:
score += 1
if "##" in content:
score += 1
if "http" in content:
score += 1
if content.count("来源") >= 2:
score += 1
if content.count("评级") >= 1 or content.count("质量") >= 1:
score += 1
quality_score = (score / total_checks) * 100
quality_scores.append((inst_file.name, quality_score))
except Exception as e:
self.warnings.append(f"无法读取机构档案 {inst_file.name}: {str(e)}")
if quality_scores:
avg_score = sum(s for _, s in quality_scores) / len(quality_scores)
high_quality = sum(1 for _, s in quality_scores if s >= 90)
medium_quality = sum(1 for _, s in quality_scores if 80 <= s < 90)
print(f" 平均质量得分: {avg_score:.1f}%")
print(f" 高质量档案 (≥90%): {high_quality}")
print(f" 中等质量档案 (80-90%): {medium_quality}")
if avg_score < 80:
self.issues.append(f"机构档案平均质量低于80%: {avg_score:.1f}%")
else:
self.successes.append(f"机构档案质量良好: {avg_score:.1f}%")
return len(inst_files), quality_scores
return 0, []
def verify_literature_integration(self) -> Tuple[int, int]:
print("\n📖 检查文献库整合...")
literature_db_path = self.edu_ai_path / "文献库" / "文献索引数据库.json"
if not literature_db_path.exists():
self.issues.append("文献索引数据库不存在")
return 0, 0
try:
with open(literature_db_path, "r", encoding="utf-8") as f:
db = json.load(f)
literature_count = len(db.get("entries", []))
doi_count = sum(1 for entry in db.get("entries", []) if entry.get("doi"))
print(f" 文献总数: {literature_count}")
print(f" 包含DOI的文献: {doi_count}")
print(
f" DOI覆盖率: {(doi_count / literature_count * 100):.1f}%"
if literature_count > 0
else " 0%"
)
if doi_count / literature_count < 0.7 if literature_count > 0 else True:
self.warnings.append(f"DOI覆盖率较低: {doi_count}/{literature_count}")
else:
self.successes.append(
f"文献库DOI覆盖率良好: {doi_count}/{literature_count}"
)
return literature_count, doi_count
except Exception as e:
self.issues.append(f"无法读取文献数据库: {str(e)}")
return 0, 0
def verify_cross_document_references(self) -> int:
print("\n🔗 检查跨文档引用...")
total_wikilinks = 0
total_files = 0
for md_file in self.edu_ai_path.rglob("*.md"):
try:
content = md_file.read_text(encoding="utf-8")
wikilinks = content.count("[[") - content.count("[![")
total_wikilinks += wikilinks
total_files += 1
except Exception:
pass
avg_links = total_wikilinks / total_files if total_files > 0 else 0
print(f" 扫描文件数: {total_files}")
print(f" Wiki链接总数: {total_wikilinks}")
print(f" 平均每文件链接数: {avg_links:.1f}")
if avg_links < 2:
self.warnings.append(f"跨文档引用较少,平均{avg_links:.1f}个/文件")
else:
self.successes.append(f"跨文档引用正常: {avg_links:.1f}个/文件")
return total_wikilinks
def generate_health_report(self) -> Dict:
print("\n📊 生成知识库健康报告...")
report = {
"timestamp": datetime.datetime.now().isoformat(),
"summary": {
"total_issues": len(self.issues),
"total_warnings": len(self.warnings),
"total_successes": len(self.successes),
"overall_health": "健康" if len(self.issues) == 0 else "需要修复",
},
"issues": self.issues,
"warnings": self.warnings,
"successes": self.successes,
"recommendations": [],
}
if len(self.issues) > 0:
report["recommendations"].append("优先修复所有问题,确保自动化机制正常运行")
if len(self.warnings) > 0:
report["recommendations"].append("审查警告项,优化知识库质量")
if report["summary"]["overall_health"] == "健康":
report["recommendations"].append("保持当前质量,定期执行同步验证")
return report
def save_health_report(self, report: Dict) -> Path:
output_dir = self.edu_ai_path / "知识库同步"
output_dir.mkdir(parents=True, exist_ok=True)
today = datetime.date.today()
week_number = today.isocalendar()[1]
year = today.year
report_file = output_dir / f"{year}-W{week_number}-同步验证报告.md"
md_content = f"""# 知识库同步验证报告
## 基本信息
- **验证时间**: {report["timestamp"]}
- **报告编号**: {year}-W{week_number}
- **项目名称**: 教育AI研究
## 总体健康状态
| 指标 | 值 |
|------|-----|
| 问题数量 | {report["summary"]["total_issues"]} |
| 警告数量 | {report["summary"]["total_warnings"]} |
| 成功项 | {report["summary"]["total_successes"]} |
| 总体状态 | {"🟢 健康" if report["summary"]["overall_health"] == "健康" else "🔴 需要修复"} |
## 详细验证结果
### ✅ 成功项
"""
for i, success in enumerate(report["successes"], 1):
md_content += f"{i}. {success}\n"
md_content += f"""
### ⚠️ 警告项
"""
for i, warning in enumerate(report["warnings"], 1):
md_content += f"{i}. {warning}\n"
md_content += f"""
### ❌ 问题项
"""
for i, issue in enumerate(report["issues"], 1):
md_content += f"{i}. {issue}\n"
md_content += f"""
## 改进建议
"""
for i, recommendation in enumerate(report["recommendations"], 1):
md_content += f"{i}. {recommendation}\n"
md_content += f"""
## 下一步行动
- [ ] 修复所有问题项
- [ ] 审查并解决警告项
- [ ] 根据建议优化知识库
- [ ] 下周再次执行验证
---
*报告生成时间: {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}*
"""
report_file.write_text(md_content, encoding="utf-8")
print(f"\n✅ 健康报告已保存: {report_file}")
return report_file
def main():
print("=" * 60)
print("教育AI研究 - 知识库自动同步验证")
print("=" * 60)
print(f"开始时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print()
vault_root = Path(__file__).parent.parent.parent.parent
edu_ai_path = vault_root / "raw" / "教育AI研究"
verifier = KnowledgeSyncVerifier(vault_root)
verifier.edu_ai_path = edu_ai_path
vault_root = Path(__file__).parent.parent.parent.parent
edu_ai_path = vault_root / "raw" / "教育AI研究"
verifier = KnowledgeSyncVerifier(vault_root)
verifier.edu_ai_path = edu_ai_path
print(f"项目根目录: {vault_root}")
print(f"教育AI研究路径: {verifier.edu_ai_path}")
print()
verifier.verify_directory_structure()
verifier.verify_knowledge_card_consistency()
verifier.verify_institution_archives()
verifier.verify_literature_integration()
verifier.verify_cross_document_references()
report = verifier.generate_health_report()
report_file = verifier.save_health_report(report)
print("\n" + "=" * 60)
print("验证完成")
print("=" * 60)
print(f"报告文件: {report_file}")
print(f"总体状态: {report['summary']['overall_health']}")
print(f"问题数: {report['summary']['total_issues']}")
print(f"警告数: {report['summary']['total_warnings']}")
print("=" * 60)
if __name__ == "__main__":
main()