folder_scanner.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415
  1. import os
  2. import logging
  3. from typing import Dict, List, Optional
  4. from dataclasses import dataclass
  5. # 配置日志
  6. logging.basicConfig(
  7. level=logging.INFO,
  8. format='%(asctime)s - %(levelname)s - %(message)s'
  9. )
  10. logger = logging.getLogger(__name__)
  11. @dataclass
  12. class ScanConfig:
  13. """扫描配置"""
  14. # 决策agent的期望数量,-1表示不校验
  15. expected_decision_count: int = -1
  16. # 大纲生成agent的期望数量,-1表示不校验
  17. expected_outline_count: int = -1
  18. # 指标计算agent的期望数量,-1表示不校验
  19. expected_metric_count: int = -1
  20. # 数据转换agent的期望数量,-1表示不校验
  21. expected_transform_count: int = -1
  22. # 期望扫描的文件夹总数(用于计算文件成功率)
  23. expected_folder_count: int = -1
  24. @dataclass
  25. class ScanResult:
  26. """单个文件夹扫描结果"""
  27. folder_name: str
  28. decision_files: List[str]
  29. outline_files: List[str]
  30. metric_files: List[str]
  31. success_metric_files: List[str]
  32. transform_files: List[str] # 新增:数据转换文件
  33. # 实际数量
  34. actual_decision_count: int
  35. actual_outline_count: int
  36. actual_metric_count: int
  37. actual_success_metric_count: int
  38. actual_transform_count: int # 新增:数据转换数量
  39. # 准确率(如果配置了期望值)
  40. decision_accuracy: Optional[float] = None
  41. outline_accuracy: Optional[float] = None
  42. metric_accuracy: Optional[float] = None
  43. transform_accuracy: Optional[float] = None # 新增:数据转换准确率
  44. class FolderScanner:
  45. """文件夹扫描工具类"""
  46. def __init__(self, base_path: str = "."):
  47. """
  48. 初始化扫描器
  49. Args:
  50. base_path: 基础路径,默认为当前目录
  51. """
  52. self.base_path = base_path
  53. self.results: List[ScanResult] = []
  54. def scan_folders(self, folder_count: int, config: Optional[ScanConfig] = None) -> Dict:
  55. """
  56. 扫描指定数量的文件夹
  57. Args:
  58. folder_count: 要扫描的文件夹数量
  59. config: 扫描配置,包含各agent的期望数量
  60. Returns:
  61. 扫描统计结果
  62. """
  63. if config is None:
  64. config = ScanConfig()
  65. logger.info(f"开始扫描 {folder_count} 个文件夹...")
  66. logger.info(f"期望配置: 决策agent={config.expected_decision_count}, "
  67. f"大纲agent={config.expected_outline_count}, "
  68. f"指标agent={config.expected_metric_count}, "
  69. f"数据转换agent={config.expected_transform_count}")
  70. self.results = []
  71. missing_folders = [] # 记录缺失的文件夹
  72. for i in range(1, folder_count + 1):
  73. folder_name = f"api_results_{i}"
  74. folder_path = os.path.join(self.base_path, folder_name)
  75. if not os.path.exists(folder_path):
  76. logger.warning(f"文件夹不存在: {folder_path}")
  77. missing_folders.append(folder_name)
  78. continue
  79. result = self._scan_single_folder(folder_name, folder_path, config)
  80. self.results.append(result)
  81. return self._generate_report(config, folder_count, missing_folders)
  82. def _calculate_accuracy(self, actual: int, expected: int) -> float:
  83. """计算准确率"""
  84. if actual == expected:
  85. return 100.0
  86. elif actual < expected:
  87. # 实际少于期望
  88. return (actual / expected) * 100
  89. else:
  90. # 实际多于期望,也应该计算偏差
  91. # 比如期望3个,实际4个,准确率 = 3/4 * 100 = 75%
  92. return (expected / actual) * 100
  93. def _scan_single_folder(self, folder_name: str, folder_path: str, config: ScanConfig) -> ScanResult:
  94. """扫描单个文件夹"""
  95. all_files = os.listdir(folder_path)
  96. # 分类文件
  97. decision_files = []
  98. outline_files = []
  99. metric_files = []
  100. success_metric_files = []
  101. transform_files = [] # 新增:数据转换文件
  102. for file in all_files:
  103. if file.endswith('.md'):
  104. if '规划决策' in file:
  105. decision_files.append(file)
  106. elif '大纲生成' in file:
  107. outline_files.append(file)
  108. elif '数据转换' in file: # 新增:识别数据转换文件
  109. transform_files.append(file)
  110. elif file.endswith('.json'):
  111. metric_files.append(file)
  112. if '_success.json' in file:
  113. success_metric_files.append(file)
  114. # 计算实际数量
  115. actual_decision_count = len(decision_files)
  116. actual_outline_count = len(outline_files)
  117. actual_metric_count = len(metric_files)
  118. actual_success_metric_count = len(success_metric_files)
  119. actual_transform_count = len(transform_files) # 新增:数据转换数量
  120. # 计算准确率(如果配置了期望值)
  121. decision_accuracy = None
  122. outline_accuracy = None
  123. metric_accuracy = None
  124. transform_accuracy = None # 新增:数据转换准确率
  125. if config.expected_decision_count != -1 and config.expected_decision_count > 0:
  126. decision_accuracy = self._calculate_accuracy(actual_decision_count, config.expected_decision_count)
  127. if config.expected_outline_count != -1 and config.expected_outline_count > 0:
  128. outline_accuracy = self._calculate_accuracy(actual_outline_count, config.expected_outline_count)
  129. if config.expected_metric_count != -1 and config.expected_metric_count > 0:
  130. metric_accuracy = self._calculate_accuracy(actual_metric_count, config.expected_metric_count)
  131. if config.expected_transform_count != -1 and config.expected_transform_count > 0:
  132. transform_accuracy = self._calculate_accuracy(actual_transform_count, config.expected_transform_count)
  133. return ScanResult(
  134. folder_name=folder_name,
  135. decision_files=decision_files,
  136. outline_files=outline_files,
  137. metric_files=metric_files,
  138. success_metric_files=success_metric_files,
  139. transform_files=transform_files, # 新增
  140. actual_decision_count=actual_decision_count,
  141. actual_outline_count=actual_outline_count,
  142. actual_metric_count=actual_metric_count,
  143. actual_success_metric_count=actual_success_metric_count,
  144. actual_transform_count=actual_transform_count, # 新增
  145. decision_accuracy=decision_accuracy,
  146. outline_accuracy=outline_accuracy,
  147. metric_accuracy=metric_accuracy,
  148. transform_accuracy=transform_accuracy # 新增
  149. )
  150. def _generate_report(self, config: ScanConfig, expected_folder_count: int, missing_folders: List[str]) -> Dict:
  151. """生成报告"""
  152. total_folders = len(self.results)
  153. # 计算文件成功率
  154. file_success_rate = 0
  155. if expected_folder_count > 0:
  156. file_success_rate = (total_folders / expected_folder_count) * 100
  157. # 计算平均准确率
  158. avg_decision_accuracy = 0
  159. avg_outline_accuracy = 0
  160. avg_metric_accuracy = 0
  161. avg_transform_accuracy = 0 # 新增:数据转换平均准确率
  162. if total_folders > 0:
  163. # 决策agent平均准确率
  164. decision_accuracies = [r.decision_accuracy for r in self.results if r.decision_accuracy is not None]
  165. if decision_accuracies:
  166. avg_decision_accuracy = sum(decision_accuracies) / len(decision_accuracies)
  167. # 大纲agent平均准确率
  168. outline_accuracies = [r.outline_accuracy for r in self.results if r.outline_accuracy is not None]
  169. if outline_accuracies:
  170. avg_outline_accuracy = sum(outline_accuracies) / len(outline_accuracies)
  171. # 指标agent平均准确率
  172. metric_accuracies = [r.metric_accuracy for r in self.results if r.metric_accuracy is not None]
  173. if metric_accuracies:
  174. avg_metric_accuracy = sum(metric_accuracies) / len(metric_accuracies)
  175. # 数据转换agent平均准确率
  176. transform_accuracies = [r.transform_accuracy for r in self.results if r.transform_accuracy is not None]
  177. if transform_accuracies:
  178. avg_transform_accuracy = sum(transform_accuracies) / len(transform_accuracies)
  179. # 计算综合成功率(基于所有配置的校验项)
  180. comprehensive_success_rate = 0
  181. success_items = []
  182. # 添加所有配置的准确率
  183. if config.expected_decision_count != -1:
  184. success_items.append(avg_decision_accuracy)
  185. if config.expected_outline_count != -1:
  186. success_items.append(avg_outline_accuracy)
  187. if config.expected_metric_count != -1:
  188. success_items.append(avg_metric_accuracy)
  189. if config.expected_transform_count != -1:
  190. success_items.append(avg_transform_accuracy)
  191. if success_items:
  192. comprehensive_success_rate = sum(success_items) / len(success_items)
  193. # 统计成功指标文件总数
  194. total_success_metric = sum(r.actual_success_metric_count for r in self.results)
  195. total_metric_files = sum(r.actual_metric_count for r in self.results)
  196. report = {
  197. "total_folders_scanned": total_folders,
  198. "expected_folder_count": expected_folder_count,
  199. "missing_folders": missing_folders,
  200. "file_success_rate": f"{file_success_rate:.2f}%",
  201. "accuracy_statistics": {
  202. "comprehensive_success_rate": f"{comprehensive_success_rate:.2f}%", # 综合成功率
  203. "decision_accuracy": f"{avg_decision_accuracy:.2f}%" if config.expected_decision_count != -1 else "未配置",
  204. "outline_accuracy": f"{avg_outline_accuracy:.2f}%" if config.expected_outline_count != -1 else "未配置",
  205. "metric_accuracy": f"{avg_metric_accuracy:.2f}%" if config.expected_metric_count != -1 else "未配置",
  206. "transform_accuracy": f"{avg_transform_accuracy:.2f}%" if config.expected_transform_count != -1 else "未配置",
  207. },
  208. "metric_statistics": {
  209. "total_metric_files": total_metric_files,
  210. "total_success_metric_files": total_success_metric,
  211. "success_ratio": f"{(total_success_metric / total_metric_files * 100):.2f}%" if total_metric_files > 0 else "0.00%"
  212. },
  213. "expected_counts": {
  214. "decision": config.expected_decision_count,
  215. "outline": config.expected_outline_count,
  216. "metric": config.expected_metric_count,
  217. "transform": config.expected_transform_count
  218. },
  219. "details": []
  220. }
  221. # 添加详细信息
  222. for result in self.results:
  223. detail = {
  224. "folder": result.folder_name,
  225. "actual_counts": {
  226. "decision": result.actual_decision_count,
  227. "outline": result.actual_outline_count,
  228. "metric": result.actual_metric_count,
  229. "success_metric": result.actual_success_metric_count,
  230. "transform": result.actual_transform_count # 新增
  231. },
  232. "accuracies": {}
  233. }
  234. # 只添加有准确率的项
  235. if result.decision_accuracy is not None:
  236. detail["accuracies"]["decision"] = f"{result.decision_accuracy:.2f}%"
  237. if result.outline_accuracy is not None:
  238. detail["accuracies"]["outline"] = f"{result.outline_accuracy:.2f}%"
  239. if result.metric_accuracy is not None:
  240. detail["accuracies"]["metric"] = f"{result.metric_accuracy:.2f}%"
  241. if result.transform_accuracy is not None:
  242. detail["accuracies"]["transform"] = f"{result.transform_accuracy:.2f}%"
  243. report["details"].append(detail)
  244. return report
  245. def print_report(self, report: Dict):
  246. """打印报告"""
  247. logger.info("=" * 50)
  248. logger.info("文件夹扫描统计报告")
  249. logger.info("=" * 50)
  250. # 文件成功率统计
  251. logger.info(f"期望扫描文件夹数: {report['expected_folder_count']}")
  252. logger.info(f"实际扫描文件夹数: {report['total_folders_scanned']}")
  253. logger.info(f"文件成功率: {report['file_success_rate']}")
  254. if report['missing_folders']:
  255. logger.info(f"缺失文件夹: {', '.join(report['missing_folders'][:5])}")
  256. if len(report['missing_folders']) > 5:
  257. logger.info(f" ... 还有{len(report['missing_folders']) - 5}个缺失文件夹")
  258. # 准确率统计
  259. logger.info("\n📊 准确率统计:")
  260. acc_stats = report['accuracy_statistics']
  261. # 显示综合成功率
  262. logger.info(f" 综合成功率: {acc_stats['comprehensive_success_rate']}")
  263. # 只显示配置了的准确率
  264. if acc_stats['decision_accuracy'] != "未配置":
  265. logger.info(f" 决策agent准确率: {acc_stats['decision_accuracy']}")
  266. if acc_stats['outline_accuracy'] != "未配置":
  267. logger.info(f" 大纲agent准确率: {acc_stats['outline_accuracy']}")
  268. if acc_stats['metric_accuracy'] != "未配置":
  269. logger.info(f" 指标agent准确率: {acc_stats['metric_accuracy']}")
  270. if acc_stats['transform_accuracy'] != "未配置":
  271. logger.info(f" 数据转换agent准确率: {acc_stats['transform_accuracy']}")
  272. # 指标文件统计
  273. logger.info("\n📈 指标文件统计:")
  274. metric_stats = report['metric_statistics']
  275. logger.info(f" 指标文件总数: {metric_stats['total_metric_files']}")
  276. logger.info(f" 成功指标文件数: {metric_stats['total_success_metric_files']}")
  277. logger.info(f" 指标成功率: {metric_stats['success_ratio']}")
  278. # 期望值显示
  279. logger.info("\n⚙ 配置期望值:")
  280. expected = report['expected_counts']
  281. if expected['decision'] != -1:
  282. logger.info(f" 决策agent: {expected['decision']}个")
  283. if expected['outline'] != -1:
  284. logger.info(f" 大纲agent: {expected['outline']}个")
  285. if expected['metric'] != -1:
  286. logger.info(f" 指标agent: {expected['metric']}个")
  287. if expected['transform'] != -1:
  288. logger.info(f" 数据转换agent: {expected['transform']}个")
  289. logger.info("=" * 50)
  290. # 打印前3个文件夹的详情(避免日志太长)
  291. if report['details']:
  292. logger.info("前3个文件夹详情:")
  293. for i, detail in enumerate(report['details'][:50]):
  294. logger.info(f"\n 📂 {detail['folder']}:")
  295. actual = detail['actual_counts']
  296. # 构建实际数量字符串
  297. count_parts = []
  298. if expected['decision'] != -1:
  299. count_parts.append(f"决策:{actual['decision']}")
  300. if expected['outline'] != -1:
  301. count_parts.append(f"大纲:{actual['outline']}")
  302. if expected['metric'] != -1:
  303. count_parts.append(f"指标:{actual['metric']}")
  304. if expected['transform'] != -1:
  305. count_parts.append(f"数据转换:{actual['transform']}")
  306. count_parts.append(f"成功指标:{actual['success_metric']}")
  307. logger.info(f" 实际数量 - {', '.join(count_parts)}")
  308. # 构建准确率字符串
  309. if detail['accuracies']:
  310. accuracy_parts = []
  311. for key, value in detail['accuracies'].items():
  312. if key == 'decision':
  313. accuracy_parts.append(f"决策:{value}")
  314. elif key == 'outline':
  315. accuracy_parts.append(f"大纲:{value}")
  316. elif key == 'metric':
  317. accuracy_parts.append(f"指标:{value}")
  318. elif key == 'transform':
  319. accuracy_parts.append(f"数据转换:{value}")
  320. logger.info(f" 准确率 - {', '.join(accuracy_parts)}")
  321. if len(report['details']) > 50:
  322. logger.info(f" ... 还有{len(report['details']) - 3}个文件夹未显示")
  323. logger.info("=" * 50)
  324. # 使用示例
  325. def main():
  326. """使用示例"""
  327. # 创建扫描器
  328. scanner = FolderScanner(base_path=".")
  329. # 配置期望值
  330. config = ScanConfig(
  331. expected_decision_count=-1, # 期望每个文件夹有3个决策agent文件
  332. expected_outline_count=1, # 期望每个文件夹有1个大纲生成agent文件
  333. expected_metric_count=19, # 期望每个文件夹有20个指标计算agent文件
  334. expected_transform_count=2, # 期望每个文件夹有2个数据转换agent文件
  335. expected_folder_count=10 # 期望扫描100个文件夹
  336. )
  337. # 扫描100个文件夹
  338. report = scanner.scan_folders(folder_count=10, config=config)
  339. # 打印报告
  340. scanner.print_report(report)
  341. # 也可以返回报告数据供进一步处理
  342. return report
  343. if __name__ == "__main__":
  344. main()