466 lines
17 KiB
Python
466 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
GUI客户端版本创建器
|
||
⚠️ 仅用于CI/CD环境（Gitea Actions），不打包到用户端GUI
|
||
|
||
安全说明：
|
||
- 本脚本包含数据库凭证，仅在受控的CI/CD环境运行
|
||
- 用户端GUI不包含此脚本，避免数据库凭证泄漏
|
||
- 用户端GUI只读取本地version_history.json文件
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import logging
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, Optional
|
||
|
||
# 添加项目根目录到路径
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s [%(levelname)s] %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 数据库配置（与后端一致）
|
||
# ⚠️ 仅在CI/CD环境使用，不会打包到用户端
|
||
DB_CONFIG = {
|
||
'host': os.getenv('DB_HOST', '8.155.9.53'),
|
||
'port': int(os.getenv('DB_PORT', '5400')),
|
||
'database': os.getenv('DB_NAME', 'ai_web'),
|
||
'user': os.getenv('DB_USER', 'user_emKCAb'),
|
||
'password': os.getenv('DB_PASSWORD', 'password_ee2iQ3')
|
||
}
|
||
|
||
|
||
class GitUtils:
|
||
"""Git操作工具"""
|
||
|
||
@staticmethod
|
||
def run_command(args: list) -> Optional[str]:
|
||
"""执行Git命令"""
|
||
import subprocess
|
||
try:
|
||
result = subprocess.run(
|
||
['git'] + args,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8'
|
||
)
|
||
return result.stdout.strip() if result.returncode == 0 else None
|
||
except Exception as e:
|
||
logger.error(f"Git命令执行失败: {e}")
|
||
return None
|
||
|
||
@classmethod
|
||
def get_commit_info(cls) -> Dict[str, str]:
|
||
"""获取当前提交信息"""
|
||
return {
|
||
'message': cls.run_command(['log', '-1', '--pretty=format:%B']) or "自动发布",
|
||
'hash': cls.run_command(['rev-parse', 'HEAD']) or "",
|
||
'short_hash': (cls.run_command(['rev-parse', 'HEAD']) or "")[:8],
|
||
'author': cls.run_command(['log', '-1', '--pretty=format:%an']) or "系统",
|
||
'commit_date': cls.run_command(['log', '-1', '--pretty=format:%ci']) or "",
|
||
'branch': cls.run_command(['rev-parse', '--abbrev-ref', 'HEAD']) or "unknown"
|
||
}
|
||
|
||
@classmethod
|
||
def get_commit_stats(cls) -> Dict[str, int]:
|
||
"""获取提交统计"""
|
||
stats_output = cls.run_command(['diff', '--numstat', 'HEAD~1', 'HEAD'])
|
||
if not stats_output:
|
||
return {'files_changed': 0, 'lines_added': 0, 'lines_deleted': 0}
|
||
|
||
files_changed = 0
|
||
lines_added = 0
|
||
lines_deleted = 0
|
||
|
||
for line in stats_output.split('\n'):
|
||
if line.strip():
|
||
parts = line.split('\t')
|
||
if len(parts) >= 2:
|
||
try:
|
||
added = int(parts[0]) if parts[0].isdigit() else 0
|
||
deleted = int(parts[1]) if parts[1].isdigit() else 0
|
||
files_changed += 1
|
||
lines_added += added
|
||
lines_deleted += deleted
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
return {
|
||
'files_changed': files_changed,
|
||
'lines_added': lines_added,
|
||
'lines_deleted': lines_deleted
|
||
}
|
||
|
||
|
||
class DatabaseVersionManager:
|
||
"""
|
||
数据库版本管理器
|
||
⚠️ 安全警告：仅在CI/CD环境使用
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.project_root = PROJECT_ROOT
|
||
self.version_file = self.project_root / 'version_history.json'
|
||
self.config_file = self.project_root / 'config.py'
|
||
self.git_utils = GitUtils()
|
||
self.db_config = DB_CONFIG
|
||
self.conn = None
|
||
|
||
def connect_db(self):
|
||
"""连接数据库"""
|
||
try:
|
||
import psycopg2
|
||
self.conn = psycopg2.connect(**self.db_config)
|
||
logger.info("✅ 数据库连接成功")
|
||
return True
|
||
except ImportError:
|
||
logger.error("❌ psycopg2未安装，请执行: pip install psycopg2-binary")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"❌ 数据库连接失败: {e}")
|
||
return False
|
||
|
||
def close_db(self):
|
||
"""关闭数据库连接"""
|
||
if self.conn:
|
||
self.conn.close()
|
||
logger.info("数据库连接已关闭")
|
||
|
||
def analyze_update_type(self, commit_message: str) -> str:
|
||
"""分析更新类型"""
|
||
message = commit_message.lower()
|
||
|
||
# 手动标记优先
|
||
if '[major]' in message or '【major】' in message:
|
||
return 'major'
|
||
elif '[minor]' in message or '【minor】' in message:
|
||
return 'minor'
|
||
elif '[patch]' in message or '【patch】' in message:
|
||
return 'patch'
|
||
|
||
# 关键词识别
|
||
major_keywords = ['重构', 'refactor', '架构', 'framework', 'breaking', '大版本', '底层重写']
|
||
minor_keywords = ['新增', 'add', 'feature', '功能', 'feat:', 'new', '增加', 'implement', '实现']
|
||
patch_keywords = ['修复', 'fix', 'bug', '优化', 'optimize', 'improve', '调整', 'update', '界面', 'ui', '样式']
|
||
|
||
if any(keyword in message for keyword in major_keywords):
|
||
return 'major'
|
||
elif any(keyword in message for keyword in minor_keywords):
|
||
return 'minor'
|
||
elif any(keyword in message for keyword in patch_keywords):
|
||
return 'patch'
|
||
else:
|
||
return 'patch'
|
||
|
||
def get_latest_version_from_db(self) -> str:
|
||
"""从数据库获取最新版本"""
|
||
try:
|
||
cursor = self.conn.cursor()
|
||
cursor.execute("""
|
||
SELECT version FROM web_version_history
|
||
WHERE type = '水滴智能通讯插件' AND is_delete = FALSE
|
||
ORDER BY release_time DESC LIMIT 1
|
||
""")
|
||
result = cursor.fetchone()
|
||
cursor.close()
|
||
|
||
if result:
|
||
logger.info(f"📡 从数据库获取最新版本: v{result[0]}")
|
||
return result[0]
|
||
return "1.0.0"
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 从数据库获取版本失败: {e}，使用默认版本")
|
||
return "1.0.0"
|
||
|
||
def calculate_next_version(self, update_type: str) -> str:
|
||
"""计算下一个版本号"""
|
||
current_version = self.get_latest_version_from_db()
|
||
|
||
try:
|
||
parts = current_version.split('.')
|
||
major = int(parts[0]) if len(parts) > 0 else 1
|
||
minor = int(parts[1]) if len(parts) > 1 else 0
|
||
patch = int(parts[2]) if len(parts) > 2 else 0
|
||
except (ValueError, IndexError):
|
||
return "1.0.0"
|
||
|
||
if update_type == 'major':
|
||
return f"{major + 1}.0.0"
|
||
elif update_type == 'minor':
|
||
return f"{major}.{minor + 1}.0"
|
||
else: # patch
|
||
return f"{major}.{minor}.{patch + 1}"
|
||
|
||
def check_duplicate_in_db(self, commit_hash: str) -> bool:
|
||
"""检查数据库中是否已存在该版本"""
|
||
if not commit_hash:
|
||
return False
|
||
|
||
commit_id = commit_hash[:16]
|
||
|
||
try:
|
||
cursor = self.conn.cursor()
|
||
cursor.execute("""
|
||
SELECT version FROM web_version_history
|
||
WHERE content LIKE %s AND type = '水滴智能通讯插件'
|
||
ORDER BY release_time DESC LIMIT 1
|
||
""", (f'%{commit_id}%',))
|
||
result = cursor.fetchone()
|
||
cursor.close()
|
||
|
||
if result:
|
||
logger.info(f"📡 数据库检测到重复版本: v{result[0]}")
|
||
return True
|
||
return False
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 检查重复版本失败: {e}")
|
||
return False
|
||
|
||
def save_to_database(self, version_record: dict) -> bool:
|
||
"""保存版本记录到数据库（与后端完全一致）"""
|
||
try:
|
||
# 时间处理：与后端保持一致
|
||
from datetime import timezone as dt_timezone
|
||
beijing_time_naive = datetime.now()
|
||
beijing_time_as_utc = beijing_time_naive.replace(tzinfo=dt_timezone.utc)
|
||
|
||
cursor = self.conn.cursor()
|
||
cursor.execute("""
|
||
INSERT INTO web_version_history
|
||
(version, type, content, download_url, release_time, is_delete)
|
||
VALUES (%s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""", (
|
||
version_record['version'],
|
||
'水滴智能通讯插件',
|
||
version_record['content'],
|
||
version_record.get('download_url', ''),
|
||
beijing_time_as_utc,
|
||
False
|
||
))
|
||
|
||
version_id = cursor.fetchone()[0]
|
||
self.conn.commit()
|
||
cursor.close()
|
||
|
||
logger.info(f"✅ 版本记录已保存到数据库 (ID: {version_id})")
|
||
return True
|
||
|
||
except Exception as e:
|
||
self.conn.rollback()
|
||
logger.error(f"❌ 保存到数据库失败: {e}")
|
||
import traceback
|
||
logger.error(traceback.format_exc())
|
||
return False
|
||
|
||
def save_local_backup(self, version_record: dict):
|
||
"""
|
||
保存本地JSON备份
|
||
⚠️ 此文件会被打包到用户端GUI，用于版本历史查看
|
||
"""
|
||
try:
|
||
# 读取现有历史
|
||
history = []
|
||
if self.version_file.exists():
|
||
with open(self.version_file, 'r', encoding='utf-8') as f:
|
||
history = json.load(f)
|
||
|
||
# 添加新记录
|
||
history.insert(0, version_record)
|
||
|
||
# 保留最近50个版本
|
||
if len(history) > 50:
|
||
history = history[:50]
|
||
|
||
# 保存
|
||
with open(self.version_file, 'w', encoding='utf-8') as f:
|
||
json.dump(history, f, ensure_ascii=False, indent=2)
|
||
|
||
logger.info(f"✅ 本地备份已保存: {self.version_file}")
|
||
logger.info(f" 此文件将打包到用户端GUI，用于版本历史查看")
|
||
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 保存本地备份失败: {e}")
|
||
|
||
def update_config_version(self, new_version: str):
|
||
"""更新config.py中的APP_VERSION"""
|
||
if not self.config_file.exists():
|
||
logger.warning(f"配置文件不存在: {self.config_file}")
|
||
return False
|
||
|
||
try:
|
||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
import re
|
||
pattern = r'APP_VERSION\s*=\s*["\'][\d.]+["\']'
|
||
replacement = f'APP_VERSION = "{new_version}"'
|
||
|
||
if re.search(pattern, content):
|
||
new_content = re.sub(pattern, replacement, content)
|
||
|
||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||
f.write(new_content)
|
||
|
||
logger.info(f"✅ 已更新 config.py: APP_VERSION = \"{new_version}\"")
|
||
return True
|
||
else:
|
||
logger.warning("未找到APP_VERSION配置项")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新config.py失败: {e}")
|
||
return False
|
||
|
||
def create_version_record(self) -> dict:
|
||
"""
|
||
创建版本记录（直接数据库操作）
|
||
⚠️ 仅在CI/CD环境运行
|
||
"""
|
||
try:
|
||
logger.info("=" * 70)
|
||
logger.info("🚀 开始创建GUI版本记录（CI/CD环境）")
|
||
logger.info("=" * 70)
|
||
|
||
# 1. 连接数据库
|
||
if not self.connect_db():
|
||
return {'success': False, 'error': '数据库连接失败'}
|
||
|
||
# 2. 获取Git提交信息
|
||
commit_info = self.git_utils.get_commit_info()
|
||
logger.info(f"📝 提交消息: {commit_info['message'][:100]}")
|
||
logger.info(f"👤 提交作者: {commit_info['author']}")
|
||
logger.info(f"🔖 提交哈希: {commit_info['short_hash']}")
|
||
|
||
# 3. 检查是否重复
|
||
if self.check_duplicate_in_db(commit_info['hash']):
|
||
logger.info(f"⏭️ 版本记录已存在，跳过创建")
|
||
self.close_db()
|
||
return {
|
||
'success': True,
|
||
'duplicate': True,
|
||
'message': '版本记录已存在'
|
||
}
|
||
|
||
# 4. 分析版本信息
|
||
update_type = self.analyze_update_type(commit_info['message'])
|
||
next_version = self.calculate_next_version(update_type)
|
||
|
||
logger.info(f"📊 更新类型: {update_type.upper()}")
|
||
logger.info(f"🔢 新版本号: v{next_version}")
|
||
|
||
# 5. 获取提交统计
|
||
stats = self.git_utils.get_commit_stats()
|
||
logger.info(f"📈 代码统计: {stats['files_changed']} 文件, "
|
||
f"+{stats['lines_added']}/-{stats['lines_deleted']} 行")
|
||
|
||
# 6. 创建版本记录
|
||
version_record = {
|
||
'version': next_version,
|
||
'update_type': update_type,
|
||
'content': commit_info['message'],
|
||
'author': commit_info['author'],
|
||
'commit_hash': commit_info['hash'],
|
||
'commit_short_hash': commit_info['short_hash'],
|
||
'branch': commit_info['branch'],
|
||
'release_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
'stats': stats
|
||
}
|
||
|
||
# 7. 保存到数据库（主要存储，CI/CD专用）
|
||
logger.info("")
|
||
logger.info("💾 步骤1: 保存到PostgreSQL数据库...")
|
||
db_success = self.save_to_database(version_record)
|
||
|
||
# 8. 保存到本地JSON（用户端可见）
|
||
logger.info("💾 步骤2: 保存到本地JSON备份（用户端可见）...")
|
||
self.save_local_backup(version_record)
|
||
|
||
# 9. 更新config.py
|
||
logger.info("💾 步骤3: 更新config.py...")
|
||
self.update_config_version(next_version)
|
||
|
||
# 10. 关闭数据库连接
|
||
self.close_db()
|
||
|
||
# 11. 返回结果
|
||
logger.info("")
|
||
logger.info("=" * 70)
|
||
logger.info("✅ GUI版本记录创建完成")
|
||
logger.info("=" * 70)
|
||
logger.info(f"📦 版本号: v{next_version}")
|
||
logger.info(f"📂 类型: {update_type.upper()}")
|
||
logger.info(f"👤 作者: {commit_info['author']}")
|
||
logger.info(f"📈 变更: {stats['files_changed']} 文件, "
|
||
f"+{stats['lines_added']}/-{stats['lines_deleted']} 行")
|
||
logger.info(f"💾 数据库: {'✅ 成功' if db_success else '❌ 失败'}")
|
||
logger.info(f"💾 本地备份: ✅ 成功")
|
||
logger.info(f"💾 config.py: ✅ 成功")
|
||
logger.info("=" * 70)
|
||
logger.info("🔒 安全提示：本脚本仅在CI/CD环境运行，不会打包到用户端")
|
||
logger.info("=" * 70)
|
||
|
||
return {
|
||
'success': True,
|
||
'duplicate': False,
|
||
'version': next_version,
|
||
'update_type': update_type,
|
||
'author': commit_info['author'],
|
||
'content': commit_info['message'],
|
||
'stats': stats,
|
||
'db_saved': db_success
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 创建版本记录失败: {e}")
|
||
import traceback
|
||
logger.error(traceback.format_exc())
|
||
|
||
if self.conn:
|
||
self.close_db()
|
||
|
||
return {
|
||
'success': False,
|
||
'error': str(e)
|
||
}
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
try:
|
||
logger.info("🔒 安全检查：此脚本应仅在CI/CD环境运行")
|
||
logger.info(" 用户端GUI不应包含此脚本")
|
||
logger.info("")
|
||
|
||
manager = DatabaseVersionManager()
|
||
result = manager.create_version_record()
|
||
|
||
# 输出JSON结果供CI/CD使用
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
|
||
# 设置退出码
|
||
if not result['success']:
|
||
sys.exit(1)
|
||
|
||
except Exception as e:
|
||
logger.error(f"脚本执行失败: {e}")
|
||
print(json.dumps({
|
||
'success': False,
|
||
'error': str(e)
|
||
}, ensure_ascii=False))
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|