feat(project): move edge_inference_service contents to root and update paths
- Moved all project files and directories (config, core, models, etc.) from edge_inference_service/ to the repository root ai_edge/ - Updated model path in config/settings.py to reflect new structure - Revised usage paths in __init__.py documentation
This commit is contained in:
256
utils/version_control.py
Normal file
256
utils/version_control.py
Normal file
@@ -0,0 +1,256 @@
|
||||
"""
|
||||
版本控制模块
|
||||
记录代码更新历史，包括更新时间、内容、修改人及影响范围
|
||||
"""
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user