提交ci/cd文件
This commit is contained in:
Gitea Actions
469
.gitea/scripts/gui_version_creator.py
Normal file
469
.gitea/scripts/gui_version_creator.py
Normal file
@@ -0,0 +1,469 @@
|
||||
#!/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
|
||||
import uuid
|
||||
|
||||
beijing_time_naive = datetime.now()
|
||||
beijing_time_as_utc = beijing_time_naive.replace(tzinfo=dt_timezone.utc)
|
||||
|
||||
# 生成UUID
|
||||
record_id = str(uuid.uuid4())
|
||||
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO web_version_history
|
||||
(id, version, type, content, download_url, release_time, is_delete)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||||
""", (
|
||||
record_id,
|
||||
version_record['version'],
|
||||
'水滴智能通讯插件',
|
||||
version_record['content'],
|
||||
version_record.get('download_url', ''),
|
||||
beijing_time_as_utc,
|
||||
False
|
||||
))
|
||||
|
||||
self.conn.commit()
|
||||
cursor.close()
|
||||
|
||||
logger.info(f"✅ 版本记录已保存到数据库 (ID: {record_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()
|
||||
Reference in New Issue
Block a user