257 lines
8.9 KiB
Python
257 lines
8.9 KiB
Python
|
"""
|
|||
|
|
版本控制模块
|
|||
|
|
记录代码更新历史，包括更新时间、内容、修改人及影响范围
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import json
|
|||
|
|
import logging
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import Any, Dict, List, Optional
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class VersionControl:
|
|||
|
|
"""版本控制管理类"""
|
|||
|
|
|
|||
|
|
def __init__(self, changelog_path: str = "./CHANGELOG.md"):
|
|||
|
|
self.changelog_path = changelog_path
|
|||
|
|
self.current_version = "1.0.0"
|
|||
|
|
self._init_changelog()
|
|||
|
|
|
|||
|
|
def _init_changelog(self):
|
|||
|
|
"""初始化CHANGELOG文件"""
|
|||
|
|
if not os.path.exists(self.changelog_path):
|
|||
|
|
self._create_initial_changelog()
|
|||
|
|
|
|||
|
|
def _create_initial_changelog(self):
|
|||
|
|
"""创建初始CHANGELOG"""
|
|||
|
|
header = f"""# CHANGELOG - Edge_Inference_Service
|
|||
|
|
|
|||
|
|
## 版本更新记录
|
|||
|
|
|
|||
|
|
### v{self.current_version}
|
|||
|
|
**更新时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
|
|||
|
|
**更新类型**: 初始化
|
|||
|
|
**更新人员**: 系统
|
|||
|
|
**影响范围**: 项目初始化
|
|||
|
|
|
|||
|
|
- 项目初始化创建
|
|||
|
|
- 搭建基础目录结构
|
|||
|
|
- 实现核心配置模块
|
|||
|
|
- 实现工具类模块
|
|||
|
|
|
|||
|
|
---
|
|||
|
|
*Generated by Edge_Inference_Service Version Control System*
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
os.makedirs(os.path.dirname(self.changelog_path), exist_ok=True)
|
|||
|
|
with open(self.changelog_path, 'w', encoding='utf-8') as f:
|
|||
|
|
f.write(header)
|
|||
|
|
|
|||
|
|
logger.info(f"CHANGELOG文件已创建: {self.changelog_path}")
|
|||
|
|
|
|||
|
|
def record_update(
|
|||
|
|
self,
|
|||
|
|
version: str,
|
|||
|
|
update_type: str,
|
|||
|
|
description: str,
|
|||
|
|
updated_by: str = "系统",
|
|||
|
|
affected_items: Optional[List[str]] = None,
|
|||
|
|
details: Optional[Dict[str, Any]] = None
|
|||
|
|
):
|
|||
|
|
"""
|
|||
|
|
记录代码更新
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
version: 版本号
|
|||
|
|
update_type: 更新类型 (新增/修改/修复/优化)
|
|||
|
|
description: 更新描述
|
|||
|
|
updated_by: 更新人员
|
|||
|
|
affected_items: 影响范围列表
|
|||
|
|
details: 详细信息字典
|
|||
|
|
"""
|
|||
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|||
|
|
affected = affected_items or []
|
|||
|
|
|
|||
|
|
entry = f"""
|
|||
|
|
|
|||
|
|
### v{version}
|
|||
|
|
**更新时间**: {timestamp}
|
|||
|
|
**更新类型**: {update_type}
|
|||
|
|
**更新人员**: {updated_by}
|
|||
|
|
**影响范围**: {', '.join(affected) if affected else '全局'}
|
|||
|
|
|
|||
|
|
- {description}
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
if details:
|
|||
|
|
details_str = json.dumps(details, ensure_ascii=False, indent=2)
|
|||
|
|
entry += f"\n<details>\n<summary>详细信息</summary>\n\n```json\n{details_str}\n```\n</details>\n"
|
|||
|
|
|
|||
|
|
entry += "\n---"
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with open(self.changelog_path, 'r', encoding='utf-8') as f:
|
|||
|
|
content = f.read()
|
|||
|
|
|
|||
|
|
insert_pos = content.find("\n## 版本更新记录")
|
|||
|
|
if insert_pos == -1:
|
|||
|
|
insert_pos = content.find("## 版本更新记录")
|
|||
|
|
|
|||
|
|
if insert_pos != -1:
|
|||
|
|
insert_pos = content.find("\n", insert_pos)
|
|||
|
|
new_content = content[:insert_pos] + entry + content[insert_pos:]
|
|||
|
|
else:
|
|||
|
|
new_content = content + entry
|
|||
|
|
|
|||
|
|
with open(self.changelog_path, 'w', encoding='utf-8') as f:
|
|||
|
|
f.write(new_content)
|
|||
|
|
|
|||
|
|
self.current_version = version
|
|||
|
|
logger.info(f"版本更新记录已添加: v{version}")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"记录版本更新失败: {e}")
|
|||
|
|
|
|||
|
|
def record_feature_addition(self, feature_name: str, module: str, description: str, updated_by: str = "系统"):
|
|||
|
|
"""记录功能新增"""
|
|||
|
|
version = self._bump_version("minor")
|
|||
|
|
self.record_update(
|
|||
|
|
version=version,
|
|||
|
|
update_type="新增",
|
|||
|
|
description=f"新增{module}模块: {feature_name}",
|
|||
|
|
updated_by=updated_by,
|
|||
|
|
affected_items=[module],
|
|||
|
|
details={"feature": feature_name, "module": module, "description": description}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def record_bug_fix(self, bug_description: str, module: str, fix_method: str, updated_by: str = "系统"):
|
|||
|
|
"""记录Bug修复"""
|
|||
|
|
version = self._bump_version("patch")
|
|||
|
|
self.record_update(
|
|||
|
|
version=version,
|
|||
|
|
update_type="修复",
|
|||
|
|
description=f"修复{module}模块Bug: {bug_description}",
|
|||
|
|
updated_by=updated_by,
|
|||
|
|
affected_items=[module],
|
|||
|
|
details={"bug": bug_description, "module": module, "fix_method": fix_method}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def record_optimization(self, optimization_type: str, module: str, improvement: str, updated_by: str = "系统"):
|
|||
|
|
"""记录性能优化"""
|
|||
|
|
version = self._bump_version("patch")
|
|||
|
|
self.record_update(
|
|||
|
|
version=version,
|
|||
|
|
update_type="优化",
|
|||
|
|
description=f"优化{module}模块{optimization_type}: {improvement}",
|
|||
|
|
updated_by=updated_by,
|
|||
|
|
affected_items=[module],
|
|||
|
|
details={"type": optimization_type, "module": module, "improvement": improvement}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def record_refactoring(self, module: str, reason: str, changes: List[str], updated_by: str = "系统"):
|
|||
|
|
"""记录代码重构"""
|
|||
|
|
version = self._bump_version("minor")
|
|||
|
|
self.record_update(
|
|||
|
|
version=version,
|
|||
|
|
update_type="重构",
|
|||
|
|
description=f"重构{module}模块",
|
|||
|
|
updated_by=updated_by,
|
|||
|
|
affected_items=[module],
|
|||
|
|
details={"module": module, "reason": reason, "changes": changes}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def _bump_version(self, bump_type: str) -> str:
|
|||
|
|
"""版本号递增"""
|
|||
|
|
parts = self.current_version.split(".")
|
|||
|
|
major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2])
|
|||
|
|
|
|||
|
|
if bump_type == "major":
|
|||
|
|
major += 1
|
|||
|
|
minor = 0
|
|||
|
|
patch = 0
|
|||
|
|
elif bump_type == "minor":
|
|||
|
|
minor += 1
|
|||
|
|
patch = 0
|
|||
|
|
else: # patch
|
|||
|
|
patch += 1
|
|||
|
|
|
|||
|
|
new_version = f"{major}.{minor}.{patch}"
|
|||
|
|
return new_version
|
|||
|
|
|
|||
|
|
def get_changelog_content(self) -> str:
|
|||
|
|
"""获取CHANGELOG内容"""
|
|||
|
|
try:
|
|||
|
|
if os.path.exists(self.changelog_path):
|
|||
|
|
with open(self.changelog_path, 'r', encoding='utf-8') as f:
|
|||
|
|
return f.read()
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"读取CHANGELOG失败: {e}")
|
|||
|
|
return ""
|
|||
|
|
|
|||
|
|
def get_version_history(self) -> List[Dict[str, Any]]:
|
|||
|
|
"""获取版本历史"""
|
|||
|
|
content = self.get_changelog_content()
|
|||
|
|
versions = []
|
|||
|
|
|
|||
|
|
lines = content.split("\n")
|
|||
|
|
current_version = None
|
|||
|
|
|
|||
|
|
for line in lines:
|
|||
|
|
if line.startswith("### v"):
|
|||
|
|
current_version = {
|
|||
|
|
"version": line.replace("### v", "").strip(),
|
|||
|
|
"timestamp": "",
|
|||
|
|
"update_type": "",
|
|||
|
|
"updated_by": "",
|
|||
|
|
"affected_items": [],
|
|||
|
|
"description": "",
|
|||
|
|
}
|
|||
|
|
versions.append(current_version)
|
|||
|
|
elif current_version is not None:
|
|||
|
|
if line.startswith("**更新时间**:"):
|
|||
|
|
current_version["timestamp"] = line.split(":", 1)[1].strip()
|
|||
|
|
elif line.startswith("**更新类型**:"):
|
|||
|
|
current_version["update_type"] = line.split(":", 1)[1].strip()
|
|||
|
|
elif line.startswith("**更新人员**:"):
|
|||
|
|
current_version["updated_by"] = line.split(":", 1)[1].strip()
|
|||
|
|
elif line.startswith("**影响范围**:"):
|
|||
|
|
current_version["affected_items"] = [x.strip() for x in line.split(":", 1)[1].split(",")]
|
|||
|
|
elif line.startswith("- ") and not line.startswith("- 详情"):
|
|||
|
|
current_version["description"] = line.replace("- ", "").strip()
|
|||
|
|
|
|||
|
|
return versions
|
|||
|
|
|
|||
|
|
def export_version_report(self, output_path: str = "./version_report.json"):
|
|||
|
|
"""导出版本报告"""
|
|||
|
|
history = self.get_version_history()
|
|||
|
|
|
|||
|
|
report = {
|
|||
|
|
"project": "Edge_Inference_Service",
|
|||
|
|
"current_version": self.current_version,
|
|||
|
|
"total_versions": len(history),
|
|||
|
|
"version_history": history,
|
|||
|
|
"generated_at": datetime.now().isoformat(),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with open(output_path, 'w', encoding='utf-8') as f:
|
|||
|
|
json.dump(report, f, ensure_ascii=False, indent=2)
|
|||
|
|
logger.info(f"版本报告已导出: {output_path}")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"导出版本报告失败: {e}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
_version_control_instance = None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_version_control() -> VersionControl:
|
|||
|
|
"""获取版本控制单例"""
|
|||
|
|
global _version_control_instance
|
|||
|
|
if _version_control_instance is None:
|
|||
|
|
_version_control_instance = VersionControl()
|
|||
|
|
return _version_control_instance
|