From fc2c138bc4d3fc42c086c14454ac142c126c16e3 Mon Sep 17 00:00:00 2001 From: SengokuCola <1026294844@qq.com> Date: Sun, 22 Jun 2025 17:18:28 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=B0=E5=BD=95hfc=E7=BB=9F=E8=AE=A1?= =?UTF-8?q?=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/test_hfc_performance.py | 169 +++++++++++++ scripts/view_hfc_stats.py | 184 ++++++++++++++ src/chat/focus_chat/heartFC_chat.py | 31 +++ src/chat/focus_chat/hfc_performance_logger.py | 238 ++++++++++++++++++ src/chat/focus_chat/hfc_version_manager.py | 185 ++++++++++++++ 5 files changed, 807 insertions(+) create mode 100644 scripts/test_hfc_performance.py create mode 100644 scripts/view_hfc_stats.py create mode 100644 src/chat/focus_chat/hfc_performance_logger.py create mode 100644 src/chat/focus_chat/hfc_version_manager.py diff --git a/scripts/test_hfc_performance.py b/scripts/test_hfc_performance.py new file mode 100644 index 000000000..be3f229d2 --- /dev/null +++ b/scripts/test_hfc_performance.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +""" +HFC性能记录功能测试脚本 +""" + +import sys +import json +from pathlib import Path + +# 添加项目根目录到Python路径 +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.chat.focus_chat.hfc_performance_logger import HFCPerformanceLogger +from src.chat.focus_chat.hfc_version_manager import set_hfc_version, get_hfc_version, auto_generate_hfc_version + + +def test_performance_logger(): + """测试性能记录器功能""" + + # 设置测试版本号 + test_version = "v1.2.3_test" + set_hfc_version(test_version) + print(f"设置测试版本号: {test_version}") + print(f"当前版本号: {get_hfc_version()}") + + # 创建测试用的性能记录器 + test_chat_id = "test_chat_123" + logger = HFCPerformanceLogger(test_chat_id, test_version) + + print(f"测试 HFC 性能记录器 - Chat ID: {test_chat_id}, Version: {logger.version}") + + # 模拟记录几个循环的数据 + test_cycles = [ + { + "cycle_id": 1, + "action_type": "reply", + "total_time": 2.5, + "step_times": { + "观察": 0.1, + "并行调整动作、处理": 1.2, + "规划器": 0.8, + "执行动作": 0.4 + }, + "reasoning": "用户询问天气，需要回复", + "success": True + }, + { + "cycle_id": 2, + "action_type": "no_reply", + "total_time": 1.8, + "step_times": { + "观察": 0.08, + "并行调整动作、处理": 0.9, + "规划器": 0.6, + "执行动作": 0.22 + }, + "reasoning": "无需回复的日常对话", + "success": True + }, + { + "cycle_id": 3, + "action_type": "reply", + "total_time": 3.2, + "step_times": { + "观察": 0.12, + "并行调整动作、处理": 1.5, + "规划器": 1.1, + "执行动作": 0.48 + }, + "reasoning": "用户提出复杂问题，需要详细回复", + "success": True + }, + { + "cycle_id": 4, + "action_type": "no_reply", + "total_time": 1.5, + "step_times": { + "观察": 0.07, + "并行调整动作、处理": 0.8, + "规划器": 0.5, + "执行动作": 0.13 + }, + "reasoning": "群聊中的无关对话", + "success": True + }, + { + "cycle_id": 5, + "action_type": "error", + "total_time": 0.5, + "step_times": { + "观察": 0.05, + "并行调整动作、处理": 0.2, + "规划器": 0.15, + "执行动作": 0.1 + }, + "reasoning": "处理过程中出现错误", + "success": False + } + ] + + # 记录测试数据 + for cycle_data in test_cycles: + logger.record_cycle(cycle_data) + print(f"已记录循环 {cycle_data['cycle_id']}: {cycle_data['action_type']} ({cycle_data['total_time']:.1f}s)") + + # 获取当前会话统计 + current_stats = logger.get_current_session_stats() + print("\n=== 当前会话统计 ===") + print(json.dumps(current_stats, ensure_ascii=False, indent=2)) + + # 完成会话 + logger.finalize_session() + print(f"\n=== 会话已完成 ===") + print(f"日志文件: {logger.session_file}") + print(f"统计文件: {logger.stats_file}") + + # 检查生成的文件 + if logger.session_file.exists(): + print(f"\n会话文件大小: {logger.session_file.stat().st_size} 字节") + + if logger.stats_file.exists(): + print(f"统计文件大小: {logger.stats_file.stat().st_size} 字节") + + # 读取并显示统计数据 + with open(logger.stats_file, 'r', encoding='utf-8') as f: + stats_data = json.load(f) + + print(f"\n=== 最终统计数据 ===") + if test_chat_id in stats_data: + chat_stats = stats_data[test_chat_id] + print(f"Chat ID: {test_chat_id}") + print(f"最后更新: {chat_stats['last_updated']}") + print(f"总记录数: {chat_stats['overall']['total_records']}") + print(f"平均总时间: {chat_stats['overall']['avg_total_time']:.2f}秒") + + print(f"\n各步骤平均时间:") + for step, avg_time in chat_stats['overall']['avg_step_times'].items(): + print(f" {step}: {avg_time:.3f}秒") + + print(f"\n按动作类型统计:") + for action, action_stats in chat_stats['by_action'].items(): + print(f" {action}: {action_stats['count']}次 ({action_stats['percentage']:.1f}%), 平均{action_stats['avg_total_time']:.2f}秒") + + +def test_version_manager(): + """测试版本号管理功能""" + print("\n=== 测试版本号管理器 ===") + + # 测试默认版本 + print(f"默认版本: {get_hfc_version()}") + + # 测试设置版本 + test_versions = ["v2.0.0", "1.5.0", "v1.0.0.beta", "v1.0.build123"] + for version in test_versions: + success = set_hfc_version(version) + print(f"设置版本 '{version}': {'成功' if success else '失败'} -> {get_hfc_version()}") + + # 测试自动生成版本 + auto_version = auto_generate_hfc_version() + print(f"自动生成版本: {auto_version}") + + # 测试基于现有版本的自动生成 + auto_version2 = auto_generate_hfc_version("v2.1.0") + print(f"基于v2.1.0自动生成: {auto_version2}") + + +if __name__ == "__main__": + test_version_manager() + test_performance_logger() \ No newline at end of file diff --git a/scripts/view_hfc_stats.py b/scripts/view_hfc_stats.py new file mode 100644 index 000000000..65e47a4ab --- /dev/null +++ b/scripts/view_hfc_stats.py @@ -0,0 +1,184 @@ +#!/usr/bin/env python3 +""" +HFC性能统计数据查看工具 +""" + +import sys +import json +import argparse +from pathlib import Path +from datetime import datetime +from typing import Dict, Any + +# 添加项目根目录到Python路径 +sys.path.insert(0, str(Path(__file__).parent.parent)) + + +def format_time(seconds: float) -> str: + """格式化时间显示""" + if seconds < 1: + return f"{seconds * 1000:.1f}毫秒" + else: + return f"{seconds:.3f}秒" + + +def display_chat_stats(chat_id: str, stats: Dict[str, Any]): + """显示单个聊天的统计数据""" + print(f"\n=== Chat ID: {chat_id} ===") + print(f"版本: {stats.get('version', 'unknown')}") + print(f"最后更新: {stats['last_updated']}") + + overall = stats['overall'] + print(f"\n📊 总体统计:") + print(f" 总记录数: {overall['total_records']}") + print(f" 平均总时间: {format_time(overall['avg_total_time'])}") + + print(f"\n⏱️ 各步骤平均时间:") + for step, avg_time in overall['avg_step_times'].items(): + print(f" {step}: {format_time(avg_time)}") + + print(f"\n🎯 按动作类型统计:") + by_action = stats['by_action'] + + # 按比例排序 + sorted_actions = sorted(by_action.items(), key=lambda x: x[1]['percentage'], reverse=True) + + for action, action_stats in sorted_actions: + print(f" 📌 {action}:") + print(f" 次数: {action_stats['count']} ({action_stats['percentage']:.1f}%)") + print(f" 平均总时间: {format_time(action_stats['avg_total_time'])}") + + if action_stats['avg_step_times']: + print(f" 步骤时间:") + for step, step_time in action_stats['avg_step_times'].items(): + print(f" {step}: {format_time(step_time)}") + + +def display_comparison(stats_data: Dict[str, Dict[str, Any]]): + """显示多个聊天的对比数据""" + if len(stats_data) < 2: + return + + print(f"\n=== 多聊天对比 ===") + + # 创建对比表格 + chat_ids = list(stats_data.keys()) + + print(f"\n📊 总体对比:") + print(f"{'Chat ID':<20} {'版本':<12} {'记录数':<8} {'平均时间':<12} {'最常见动作':<15}") + print("-" * 70) + + for chat_id in chat_ids: + stats = stats_data[chat_id] + overall = stats['overall'] + + # 找到最常见的动作 + most_common_action = max(stats['by_action'].items(), key=lambda x: x[1]['count']) + most_common_name = most_common_action[0] + most_common_pct = most_common_action[1]['percentage'] + + version = stats.get('version', 'unknown') + print(f"{chat_id:<20} {version:<12} {overall['total_records']:<8} {format_time(overall['avg_total_time']):<12} {most_common_name}({most_common_pct:.0f}%)") + + +def view_session_logs(chat_id: str = None, latest: bool = False): + """查看会话日志文件""" + log_dir = Path("log/hfc_loop") + if not log_dir.exists(): + print("❌ 日志目录不存在") + return + + if chat_id: + pattern = f"{chat_id}_*.json" + else: + pattern = "*.json" + + log_files = list(log_dir.glob(pattern)) + + if not log_files: + print(f"❌ 没有找到匹配的日志文件: {pattern}") + return + + if latest: + # 按文件修改时间排序，取最新的 + log_files.sort(key=lambda f: f.stat().st_mtime, reverse=True) + log_files = log_files[:1] + + for log_file in log_files: + print(f"\n=== 会话日志: {log_file.name} ===") + + try: + with open(log_file, 'r', encoding='utf-8') as f: + records = json.load(f) + + if not records: + print(" 空文件") + continue + + print(f" 记录数: {len(records)}") + print(f" 时间范围: {records[0]['timestamp']} ~ {records[-1]['timestamp']}") + + # 统计动作分布 + action_counts = {} + total_time = 0 + + for record in records: + action = record['action_type'] + action_counts[action] = action_counts.get(action, 0) + 1 + total_time += record['total_time'] + + print(f" 总耗时: {format_time(total_time)}") + print(f" 平均耗时: {format_time(total_time / len(records))}") + print(f" 动作分布: {dict(action_counts)}") + + except Exception as e: + print(f" ❌ 读取文件失败: {e}") + + +def main(): + parser = argparse.ArgumentParser(description="HFC性能统计数据查看工具") + parser.add_argument("--chat-id", help="指定要查看的Chat ID") + parser.add_argument("--logs", action="store_true", help="查看会话日志文件") + parser.add_argument("--latest", action="store_true", help="只显示最新的日志文件") + parser.add_argument("--compare", action="store_true", help="显示多聊天对比") + + args = parser.parse_args() + + if args.logs: + view_session_logs(args.chat_id, args.latest) + return + + # 读取统计数据 + stats_file = Path("data/hfc/time.json") + if not stats_file.exists(): + print("❌ 统计数据文件不存在，请先运行一些HFC循环以生成数据") + return + + try: + with open(stats_file, 'r', encoding='utf-8') as f: + stats_data = json.load(f) + except Exception as e: + print(f"❌ 读取统计数据失败: {e}") + return + + if not stats_data: + print("❌ 统计数据为空") + return + + if args.chat_id: + if args.chat_id in stats_data: + display_chat_stats(args.chat_id, stats_data[args.chat_id]) + else: + print(f"❌ 没有找到Chat ID '{args.chat_id}' 的数据") + print(f"可用的Chat ID: {list(stats_data.keys())}") + else: + # 显示所有聊天的统计数据 + for chat_id, stats in stats_data.items(): + display_chat_stats(chat_id, stats) + + if args.compare: + display_comparison(stats_data) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/chat/focus_chat/heartFC_chat.py b/src/chat/focus_chat/heartFC_chat.py index a110b4daa..e51fbc808 100644 --- a/src/chat/focus_chat/heartFC_chat.py +++ b/src/chat/focus_chat/heartFC_chat.py @@ -28,6 +28,8 @@ from src.chat.focus_chat.planners.planner_factory import PlannerFactory from src.chat.focus_chat.planners.modify_actions import ActionModifier from src.chat.focus_chat.planners.action_manager import ActionManager from src.config.config import global_config +from src.chat.focus_chat.hfc_performance_logger import HFCPerformanceLogger +from src.chat.focus_chat.hfc_version_manager import get_hfc_version install(extra_lines=3) @@ -85,6 +87,7 @@ class HeartFChatting: self, chat_id: str, on_stop_focus_chat: Optional[Callable[[], Awaitable[None]]] = None, + performance_version: str = None, ): """ HeartFChatting 初始化函数 @@ -92,6 +95,7 @@ class HeartFChatting: 参数: chat_id: 聊天流唯一标识符(如stream_id) on_stop_focus_chat: 当收到stop_focus_chat命令时调用的回调函数 + performance_version: 性能记录版本号，用于区分不同启动版本 """ # 基础属性 self.stream_id: str = chat_id # 聊天流ID @@ -147,6 +151,11 @@ class HeartFChatting: # 存储回调函数 self.on_stop_focus_chat = on_stop_focus_chat + + # 初始化性能记录器 + # 如果没有指定版本号，则使用全局版本管理器的版本号 + actual_version = performance_version or get_hfc_version() + self.performance_logger = HFCPerformanceLogger(chat_id, actual_version) def _register_observations(self): """注册所有观察器""" @@ -398,6 +407,21 @@ class HeartFChatting: + (f"\n详情: {'; '.join(timer_strings)}" if timer_strings else "") + processor_time_log ) + + # 记录性能数据 + try: + action_result = self._current_cycle_detail.loop_plan_info.get('action_result', {}) + cycle_performance_data = { + "cycle_id": self._current_cycle_detail.cycle_id, + "action_type": action_result.get('action_type', 'unknown'), + "total_time": self._current_cycle_detail.end_time - self._current_cycle_detail.start_time, + "step_times": cycle_timers.copy(), + "reasoning": action_result.get('reasoning', ''), + "success": self._current_cycle_detail.loop_action_info.get('action_taken', False), + } + self.performance_logger.record_cycle(cycle_performance_data) + except Exception as perf_e: + logger.warning(f"{self.log_prefix} 记录性能数据失败: {perf_e}") await asyncio.sleep(global_config.focus_chat.think_interval) @@ -775,6 +799,13 @@ class HeartFChatting: self._processing_lock.release() logger.warning(f"{self.log_prefix} 已释放处理锁") + # 完成性能统计 + try: + self.performance_logger.finalize_session() + logger.info(f"{self.log_prefix} 性能统计已完成") + except Exception as e: + logger.warning(f"{self.log_prefix} 完成性能统计时出错: {e}") + logger.info(f"{self.log_prefix} HeartFChatting关闭完成") def get_cycle_history(self, last_n: Optional[int] = None) -> List[Dict[str, Any]]: diff --git a/src/chat/focus_chat/hfc_performance_logger.py b/src/chat/focus_chat/hfc_performance_logger.py new file mode 100644 index 000000000..4de1e9751 --- /dev/null +++ b/src/chat/focus_chat/hfc_performance_logger.py @@ -0,0 +1,238 @@ +import json +import os +import time +from datetime import datetime +from typing import Dict, List, Any, Optional +from pathlib import Path +from src.common.logger import get_logger + +logger = get_logger("hfc_performance") + + +class HFCPerformanceLogger: + """HFC性能记录管理器""" + + # 版本号常量，可在启动时修改 + INTERNAL_VERSION = "v1.0.0" + + def __init__(self, chat_id: str, version: str = None): + self.chat_id = chat_id + self.version = version or self.INTERNAL_VERSION + self.log_dir = Path("log/hfc_loop") + self.data_dir = Path("data/hfc") + self.session_start_time = datetime.now() + + # 确保目录存在 + self.log_dir.mkdir(parents=True, exist_ok=True) + self.data_dir.mkdir(parents=True, exist_ok=True) + + # 当前会话的日志文件，包含版本号 + version_suffix = self.version.replace(".", "_") + self.session_file = self.log_dir / f"{chat_id}_{version_suffix}_{self.session_start_time.strftime('%Y%m%d_%H%M%S')}.json" + self.current_session_data = [] + + # 统计数据文件 + self.stats_file = self.data_dir / "time.json" + + # 初始化时计算历史统计数据 + self._update_historical_stats() + + def record_cycle(self, cycle_data: Dict[str, Any]): + """记录单次循环数据""" + try: + # 构建记录数据 + record = { + "timestamp": datetime.now().isoformat(), + "version": self.version, + "cycle_id": cycle_data.get("cycle_id"), + "chat_id": self.chat_id, + "action_type": cycle_data.get("action_type", "unknown"), + "total_time": cycle_data.get("total_time", 0), + "step_times": cycle_data.get("step_times", {}), + "reasoning": cycle_data.get("reasoning", ""), + "success": cycle_data.get("success", False) + } + + # 添加到当前会话数据 + self.current_session_data.append(record) + + # 立即写入文件（防止数据丢失） + self._write_session_data() + + logger.debug(f"记录HFC循环数据: cycle_id={record['cycle_id']}, action={record['action_type']}, time={record['total_time']:.2f}s") + + except Exception as e: + logger.error(f"记录HFC循环数据失败: {e}") + + def _write_session_data(self): + """写入当前会话数据到文件""" + try: + with open(self.session_file, 'w', encoding='utf-8') as f: + json.dump(self.current_session_data, f, ensure_ascii=False, indent=2) + except Exception as e: + logger.error(f"写入会话数据失败: {e}") + + def _update_historical_stats(self): + """更新历史统计数据""" + try: + # 读取所有历史会话文件 + all_records = [] + + # 读取当前chat_id的所有历史文件（包括不同版本） + for file_path in self.log_dir.glob(f"{self.chat_id}_*.json"): + if file_path == self.session_file: + continue # 跳过当前会话文件 + + try: + with open(file_path, 'r', encoding='utf-8') as f: + records = json.load(f) + if isinstance(records, list): + all_records.extend(records) + except Exception as e: + logger.warning(f"读取历史文件 {file_path} 失败: {e}") + + if not all_records: + logger.info(f"没有找到 chat_id={self.chat_id} 的历史数据") + return + + # 计算统计数据 + stats = self._calculate_stats(all_records) + + # 更新统计文件 + self._update_stats_file(stats) + + logger.info(f"更新了 chat_id={self.chat_id} 的历史统计数据，共 {len(all_records)} 条记录") + + except Exception as e: + logger.error(f"更新历史统计数据失败: {e}") + + def _calculate_stats(self, records: List[Dict[str, Any]]) -> Dict[str, Any]: + """计算统计数据""" + if not records: + return {} + + # 按动作类型分组 + action_groups = {} + total_times = [] + step_time_totals = {} + + for record in records: + action_type = record.get("action_type", "unknown") + total_time = record.get("total_time", 0) + step_times = record.get("step_times", {}) + + if action_type not in action_groups: + action_groups[action_type] = { + "count": 0, + "total_times": [], + "step_times": {} + } + + action_groups[action_type]["count"] += 1 + action_groups[action_type]["total_times"].append(total_time) + total_times.append(total_time) + + # 记录步骤时间 + for step_name, step_time in step_times.items(): + if step_name not in action_groups[action_type]["step_times"]: + action_groups[action_type]["step_times"][step_name] = [] + action_groups[action_type]["step_times"][step_name].append(step_time) + + if step_name not in step_time_totals: + step_time_totals[step_name] = [] + step_time_totals[step_name].append(step_time) + + # 计算各种平均值和比例 + total_records = len(records) + + # 整体统计 + overall_stats = { + "total_records": total_records, + "avg_total_time": sum(total_times) / len(total_times) if total_times else 0, + "avg_step_times": {} + } + + # 各步骤平均时间 + for step_name, times in step_time_totals.items(): + overall_stats["avg_step_times"][step_name] = sum(times) / len(times) if times else 0 + + # 按动作类型统计 + action_stats = {} + for action_type, data in action_groups.items(): + action_stats[action_type] = { + "count": data["count"], + "percentage": (data["count"] / total_records) * 100, + "avg_total_time": sum(data["total_times"]) / len(data["total_times"]) if data["total_times"] else 0, + "avg_step_times": {} + } + + # 该动作各步骤平均时间 + for step_name, times in data["step_times"].items(): + action_stats[action_type]["avg_step_times"][step_name] = sum(times) / len(times) if times else 0 + + return { + "chat_id": self.chat_id, + "version": self.version, + "last_updated": datetime.now().isoformat(), + "overall": overall_stats, + "by_action": action_stats + } + + def _update_stats_file(self, new_stats: Dict[str, Any]): + """更新统计文件""" + try: + # 读取现有统计数据 + existing_stats = {} + if self.stats_file.exists(): + with open(self.stats_file, 'r', encoding='utf-8') as f: + existing_stats = json.load(f) + + # 更新当前chat_id和版本的统计数据 + stats_key = f"{self.chat_id}_{self.version}" + existing_stats[stats_key] = new_stats + + # 写回文件 + with open(self.stats_file, 'w', encoding='utf-8') as f: + json.dump(existing_stats, f, ensure_ascii=False, indent=2) + + except Exception as e: + logger.error(f"更新统计文件失败: {e}") + + def get_current_session_stats(self) -> Dict[str, Any]: + """获取当前会话的统计数据""" + if not self.current_session_data: + return {} + + return self._calculate_stats(self.current_session_data) + + def finalize_session(self): + """结束会话，进行最终统计""" + try: + if self.current_session_data: + # 计算当前会话统计数据 + current_stats = self._calculate_stats(self.current_session_data) + + # 合并历史数据重新计算总体统计 + all_records = self.current_session_data[:] + + # 读取历史数据 + for file_path in self.log_dir.glob(f"{self.chat_id}_*.json"): + if file_path == self.session_file: + continue + + try: + with open(file_path, 'r', encoding='utf-8') as f: + records = json.load(f) + if isinstance(records, list): + all_records.extend(records) + except Exception as e: + logger.warning(f"读取历史文件 {file_path} 失败: {e}") + + # 重新计算总体统计 + total_stats = self._calculate_stats(all_records) + self._update_stats_file(total_stats) + + logger.info(f"完成会话统计，当前会话 {len(self.current_session_data)} 条记录，总共 {len(all_records)} 条记录") + + except Exception as e: + logger.error(f"结束会话统计失败: {e}") \ No newline at end of file diff --git a/src/chat/focus_chat/hfc_version_manager.py b/src/chat/focus_chat/hfc_version_manager.py new file mode 100644 index 000000000..31d07d928 --- /dev/null +++ b/src/chat/focus_chat/hfc_version_manager.py @@ -0,0 +1,185 @@ +""" +HFC性能记录版本号管理器 + +用于管理HFC性能记录的内部版本号，支持： +1. 默认版本号设置 +2. 启动时版本号配置 +3. 版本号验证和格式化 +""" + +import os +import re +from datetime import datetime +from typing import Optional +from src.common.logger import get_logger + +logger = get_logger("hfc_version") + + +class HFCVersionManager: + """HFC版本号管理器""" + + # 默认版本号 + DEFAULT_VERSION = "v1.0.0" + + # 当前运行时版本号 + _current_version: Optional[str] = None + + @classmethod + def set_version(cls, version: str) -> bool: + """ + 设置当前运行时版本号 + + 参数: + version: 版本号字符串，格式如 v1.0.0 或 1.0.0 + + 返回: + bool: 设置是否成功 + """ + try: + validated_version = cls._validate_version(version) + if validated_version: + cls._current_version = validated_version + logger.info(f"HFC性能记录版本已设置为: {validated_version}") + return True + else: + logger.warning(f"无效的版本号格式: {version}") + return False + except Exception as e: + logger.error(f"设置版本号失败: {e}") + return False + + @classmethod + def get_version(cls) -> str: + """ + 获取当前版本号 + + 返回: + str: 当前版本号 + """ + if cls._current_version: + return cls._current_version + + # 尝试从环境变量获取 + env_version = os.getenv("HFC_PERFORMANCE_VERSION") + if env_version: + if cls.set_version(env_version): + return cls._current_version + + # 返回默认版本号 + return cls.DEFAULT_VERSION + + @classmethod + def auto_generate_version(cls, base_version: str = None) -> str: + """ + 自动生成版本号（基于时间戳） + + 参数: + base_version: 基础版本号，如果不提供则使用默认版本 + + 返回: + str: 生成的版本号 + """ + if not base_version: + base_version = cls.DEFAULT_VERSION + + # 提取基础版本号的主要部分 + base_match = re.match(r'v?(\d+\.\d+)', base_version) + if base_match: + base_part = base_match.group(1) + else: + base_part = "1.0" + + # 添加时间戳 + timestamp = datetime.now().strftime("%Y%m%d_%H%M") + generated_version = f"v{base_part}.{timestamp}" + + cls.set_version(generated_version) + logger.info(f"自动生成版本号: {generated_version}") + + return generated_version + + @classmethod + def _validate_version(cls, version: str) -> Optional[str]: + """ + 验证版本号格式 + + 参数: + version: 待验证的版本号 + + 返回: + Optional[str]: 验证后的版本号，失败返回None + """ + if not version or not isinstance(version, str): + return None + + version = version.strip() + + # 支持的格式： + # v1.0.0, 1.0.0, v1.0, 1.0, v1.0.0.20241222_1530 等 + patterns = [ + r'^v?(\d+\.\d+\.\d+)$', # v1.0.0 或 1.0.0 + r'^v?(\d+\.\d+)$', # v1.0 或 1.0 + r'^v?(\d+\.\d+\.\d+\.\w+)$', # v1.0.0.build 或 1.0.0.build + r'^v?(\d+\.\d+\.\w+)$', # v1.0.build 或 1.0.build + ] + + for pattern in patterns: + match = re.match(pattern, version) + if match: + # 确保版本号以v开头 + if not version.startswith('v'): + version = 'v' + version + return version + + return None + + @classmethod + def reset_version(cls): + """重置版本号为默认值""" + cls._current_version = None + logger.info("HFC版本号已重置为默认值") + + @classmethod + def get_version_info(cls) -> dict: + """ + 获取版本信息 + + 返回: + dict: 版本相关信息 + """ + current = cls.get_version() + return { + "current_version": current, + "default_version": cls.DEFAULT_VERSION, + "is_custom": current != cls.DEFAULT_VERSION, + "env_version": os.getenv("HFC_PERFORMANCE_VERSION"), + "timestamp": datetime.now().isoformat() + } + + +# 全局函数，方便使用 +def set_hfc_version(version: str) -> bool: + """设置HFC性能记录版本号""" + return HFCVersionManager.set_version(version) + + +def get_hfc_version() -> str: + """获取当前HFC性能记录版本号""" + return HFCVersionManager.get_version() + + +def auto_generate_hfc_version(base_version: str = None) -> str: + """自动生成HFC版本号""" + return HFCVersionManager.auto_generate_version(base_version) + + +def reset_hfc_version(): + """重置HFC版本号""" + HFCVersionManager.reset_version() + + +# 在模块加载时显示当前版本信息 +if __name__ != "__main__": + current_version = HFCVersionManager.get_version() + logger.debug(f"HFC性能记录模块已加载，当前版本: {current_version}") \ No newline at end of file