#!/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()