bidmaster-cli/src/bidmaster/nodes/toc/category_manager.py
sladro 0a06aad7f5 refactor: 目录生成Agent重构 - 消除重复代码并优化架构
## 重构成果
- 新增统一组件: ChapterFactory, CategoryManager, LLMHelper
- 重构核心节点: 精简代码量30%,消除重复逻辑
- 清理无效代码: 移除残留导入和未使用方法
- 优化导入结构: 解决循环导入风险

## 新增文件
- factories.py: 统一章节创建逻辑
- category_manager.py: 整合类别相关操作
- llm_helper.py: 封装LLM调用和解析
- constants.py: 常量定义
- utils.py: 迁移指引

## 重构节点
- 所有节点使用统一的BaseNode增强方法
- 消除重复的AI响应解析、状态管理、日志记录
- 使用工厂模式统一章节创建

## 代码质量提升
- 代码行数从960行优化到949行
- 清晰的职责分离和模块化设计
- 统一的错误处理和状态管理
- 消除所有重复功能和无效代码

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-29 10:32:35 +08:00

166 lines
5.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""类别管理器
统一管理评分项类别相关的所有逻辑,包括分组、排序、匹配等。
"""
import logging
from typing import Dict, List, Any
from ...tools.parser import ScoringCriteria, DocumentChapter
from .constants import CATEGORY_NAMES, CATEGORY_ORDER
logger = logging.getLogger(__name__)
class CategoryManager:
"""类别管理器
提供统一的类别相关操作方法。
"""
@staticmethod
def group_criteria_by_category(criteria: List[ScoringCriteria]) -> Dict[str, List[ScoringCriteria]]:
"""按类别分组评分项
Args:
criteria: 评分项列表
Returns:
分组后的字典key为类别value为该类别下的评分项列表
"""
# 初始化所有预定义类别
category_groups = {category: [] for category in CATEGORY_ORDER}
# 分组
for criterion in criteria:
category_key = criterion.category.value
if category_key in category_groups:
category_groups[category_key].append(criterion)
else:
# 未知类别默认归入技术方案
category_groups["technical_solution"].append(criterion)
logger.warning(f"未知类别 {category_key},归入技术方案类别")
# 只保留有评分项的类别
filtered_groups = {k: v for k, v in category_groups.items() if v}
# 记录分组统计
for category, items in filtered_groups.items():
total_score = sum(item.max_score for item in items)
logger.info(f"类别 {category}: {len(items)}项,总分 {total_score}")
return filtered_groups
@staticmethod
def get_category_first_index(category: str, technical_criteria: List[ScoringCriteria]) -> int:
"""获取类别在原始评分项中的首次出现位置
Args:
category: 类别名称
technical_criteria: 技术评分项列表
Returns:
首次出现的索引位置未找到时返回999
"""
for i, criteria in enumerate(technical_criteria):
if criteria.category.value == category:
return i
return 999
@staticmethod
def sort_categories_by_order(category_groups: Dict[str, List[ScoringCriteria]],
technical_criteria: List[ScoringCriteria]) -> List[str]:
"""按原始出现顺序排序类别
Args:
category_groups: 类别分组字典
technical_criteria: 原始技术评分项列表
Returns:
排序后的类别键名列表
"""
return sorted(
category_groups.keys(),
key=lambda cat: CategoryManager.get_category_first_index(cat, technical_criteria)
)
@staticmethod
def extract_category_from_chapter_id(chapter_id: str) -> str:
"""从章节ID中提取类别信息
Args:
chapter_id: 章节ID格式chapter_XX_category
Returns:
类别名称,如果解析失败返回空字符串
"""
if "_" not in chapter_id:
return ""
parts = chapter_id.split("_")
if len(parts) < 3:
return ""
return "_".join(parts[2:])
@staticmethod
def find_corresponding_criteria(chapter: DocumentChapter,
technical_criteria: List[ScoringCriteria]) -> List[ScoringCriteria]:
"""查找章节对应的评分项
Args:
chapter: 章节对象
technical_criteria: 技术评分项列表
Returns:
对应的评分项列表
"""
category = CategoryManager.extract_category_from_chapter_id(chapter.id)
if not category:
return []
return [
c for c in technical_criteria
if c.category.value == category
]
@staticmethod
def format_criteria_summary(technical_criteria: List[ScoringCriteria]) -> str:
"""格式化评分项摘要用于显示
Args:
technical_criteria: 技术评分项列表
Returns:
格式化后的字符串
"""
lines = []
# 按类别分组
category_groups = {}
for criteria in technical_criteria:
category = criteria.category.value
if category not in category_groups:
category_groups[category] = []
category_groups[category].append(criteria)
# 格式化输出
for category, items in category_groups.items():
category_name = CATEGORY_NAMES.get(category, category)
lines.append(f"{category_name}】({len(items)}项):")
for item in items:
lines.append(f" - {item.item_name} ({item.max_score}分)")
return "\n".join(lines)
@staticmethod
def calculate_category_score(criteria_list: List[ScoringCriteria]) -> int:
"""计算类别总分
Args:
criteria_list: 该类别下的评分项列表
Returns:
总分值
"""
return sum(c.max_score for c in criteria_list)