zuowei/DP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20251222

Browse Source
This commit is contained in:
zuowei1216
2025-12-22 21:06:29 +08:00
parent 8ea58fe480
commit 1b19ff1b92
179 changed files with 21895 additions and 3774 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2448
AdminTool/admin_gui.py
View File

File diff suppressed because it is too large Load Diff

540
AdminTool/auto_deploy_core.py
View File

@@ -1,540 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Core 应用自动发布脚本
完全自动化构建、打包、上传服务器、更新数据库
使用方法：
python auto_deploy_core.py --version v1.0.6
python auto_deploy_core.py --version v1.0.6 --deploy # 部署到服务器
python auto_deploy_core.py --version v1.0.6 --deploy --update-db # 部署并更新数据库
"""
import os
import sys
import shutil
import subprocess
import argparse
import zipfile
import json
from pathlib import Path
from datetime import datetime
import paramiko
import pymysql
# ==================== 配置区域 ====================
PROJECT_ROOT = Path(__file__).parent.parent.absolute() # 上一级目录
DESIGNER_DIR = PROJECT_ROOT / "Designer"
DIST_CORE_DIR = DESIGNER_DIR / "dist" / "Designer" # Core 构建输出目录
DIST_SHELL_DIR = DESIGNER_DIR / "dist" / "Shell" # Shell 构建输出目录
SERVER_DIR = PROJECT_ROOT / "Server"
ARCHIVES_DIR = SERVER_DIR / "archives"
CONFIG_FILE = Path(__file__).parent / "deploy_config.json"
# 颜色输出（Windows 兼容）
try:
import colorama
colorama.init()
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BLUE = '\033[94m'
RESET = '\033[0m'
except ImportError:
GREEN = YELLOW = RED = BLUE = RESET = ''
# ==================== 工具函数 ====================
def print_step(step_num, total_steps, message):
"""打印步骤信息"""
print(f"\n{BLUE}{'='*60}{RESET}")
print(f"{GREEN}[步骤 {step_num}/{total_steps}] {message}{RESET}")
print(f"{BLUE}{'='*60}{RESET}\n")
def print_success(message):
"""打印成功信息"""
print(f"{GREEN}✓ {message}{RESET}")
def print_warning(message):
"""打印警告信息"""
print(f"{YELLOW}⚠ {message}{RESET}")
def print_error(message):
"""打印错误信息"""
print(f"{RED}✗ {message}{RESET}")
def run_command(command, cwd=None, shell=True, check=True):
"""运行命令并返回结果"""
print(f" 执行: {command}")
try:
result = subprocess.run(
command,
cwd=cwd,
shell=shell,
check=check,
capture_output=True,
text=True,
encoding='utf-8',
errors='replace'
)
if result.stdout:
print(f" 输出: {result.stdout.strip()}")
return result
except subprocess.CalledProcessError as e:
print_error(f"命令执行失败: {e}")
if e.stderr:
print(f" 错误: {e.stderr}")
if check:
sys.exit(1)
return None
# ==================== 发布步骤 ====================
def load_config():
"""加载部署配置"""
if not CONFIG_FILE.exists():
return None
try:
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print_warning(f"加载配置文件失败: {e}")
return None
def save_config(config):
"""保存部署配置"""
try:
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(config, f, ensure_ascii=False, indent=2)
print_success(f"配置已保存: {CONFIG_FILE}")
except Exception as e:
print_error(f"保存配置失败: {e}")
def step1_build_frontend():
"""步骤 1: 构建前端（Shell + Core）"""
print_step(1, 8, "构建前端（Shell + Core）")
# 检查 package.json
package_json = DESIGNER_DIR / "package.json"
if not package_json.exists():
print_error(f"未找到 package.json: {package_json}")
sys.exit(1)
# 运行构建命令
os.chdir(DESIGNER_DIR)
print(" 正在构建 Core...")
run_command("npm run build:core", cwd=DESIGNER_DIR)
print(" 正在构建 Shell...")
run_command("npm run build", cwd=DESIGNER_DIR)
# 验证输出目录
if not DIST_CORE_DIR.exists():
print_error(f"Core 构建输出目录不存在: {DIST_CORE_DIR}")
sys.exit(1)
if not DIST_SHELL_DIR.exists():
print_error(f"Shell 构建输出目录不存在: {DIST_SHELL_DIR}")
sys.exit(1)
print_success(f"构建完成")
print(f" Core: {DIST_CORE_DIR}")
print(f" Shell: {DIST_SHELL_DIR}")
def step2_package_shell_zip(version):
"""步骤 2: 打包 Shell 为 ZIP（供 CEP 扩展下载）"""
print_step(2, 8, "打包 Shell 为 ZIP")
# 生成 ZIP 文件名
zip_filename = f"shell-{version}.zip"
zip_path = DESIGNER_DIR / "dist" / zip_filename
# 如果文件已存在，先删除
if zip_path.exists():
print_warning(f"ZIP 文件已存在，删除: {zip_path}")
zip_path.unlink()
# 创建 ZIP
print(f" 创建 ZIP: {zip_path}")
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(DIST_SHELL_DIR):
for file in files:
file_path = Path(root) / file
arcname = file_path.relative_to(DIST_SHELL_DIR)
zipf.write(file_path, arcname)
# 显示文件大小
size_mb = zip_path.stat().st_size / (1024 * 1024)
print_success(f"Shell 打包完成: {zip_path}")
print(f" 文件大小: {size_mb:.2f} MB")
return zip_path
def step3_upload_to_server(version, config):
"""步骤 3: 上传到服务器"""
print_step(3, 8, "上传到服务器")
if not config:
print_warning("未配置服务器信息，跳过上传")
return
# 连接 SSH
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
print(f" 连接服务器: {config['host']}")
ssh.connect(
hostname=config['host'],
port=int(config.get('port', 22)),
username=config['username'],
password=config.get('password', ''),
timeout=30
)
print_success("SSH 连接成功")
sftp = ssh.open_sftp()
remote_path = config['remote_path']
# 1. 创建远程目录
print(" 创建远程目录...")
commands = [
f"mkdir -p {remote_path}/shell",
f"mkdir -p {remote_path}/core/{version}",
f"mkdir -p {remote_path}/downloads"
]
for cmd in commands:
stdin, stdout, stderr = ssh.exec_command(cmd)
stdout.channel.recv_exit_status()
print_success("目录创建完成")
# 2. 上传 Shell（在线登录页）
print(" 上传 Shell（在线登录页）...")
upload_directory(sftp, DIST_SHELL_DIR, f"{remote_path}/shell")
print_success("Shell 上传完成")
# 3. 上传 Core
print(" 上传 Core（核心应用）...")
upload_directory(sftp, DIST_CORE_DIR, f"{remote_path}/core/{version}")
print_success("Core 上传完成")
# 4. 上传 Shell.zip
print(" 上传 Shell.zip（CEP 扩展下载）...")
shell_zip = DESIGNER_DIR / "dist" / f"shell-{version}.zip"
if shell_zip.exists():
remote_zip = f"{remote_path}/downloads/shell-{version}.zip"
sftp.put(str(shell_zip), remote_zip)
print_success(f"Shell.zip 上传完成: {remote_zip}")
sftp.close()
ssh.close()
print_success("所有文件上传完成")
print(f"\n访问地址:")
print(f" Shell: https://{config['host']}/shell/")
print(f" Core: https://{config['host']}/core/{version}/")
print(f" 下载: https://{config['host']}/downloads/shell-{version}.zip")
except Exception as e:
print_error(f"上传失败: {e}")
raise
finally:
ssh.close()
def upload_directory(sftp, local_dir, remote_dir):
"""递归上传目录"""
if not os.path.exists(local_dir):
raise Exception(f"本地目录不存在: {local_dir}")
# 创建远程目录
try:
sftp.stat(remote_dir)
except IOError:
sftp.mkdir(remote_dir)
# 递归上传文件
for item in os.listdir(local_dir):
local_path = os.path.join(local_dir, item)
remote_path = remote_dir + '/' + item
if os.path.isfile(local_path):
print(f" → {item}")
sftp.put(local_path, remote_path)
elif os.path.isdir(local_path):
upload_directory(sftp, local_path, remote_path)
def step4_update_mysql(version, config):
"""步骤 4: 更新 MySQL 数据库 (通过 SSH 执行 Docker 命令)"""
print_step(4, 8, "更新 MySQL 数据库")
# 注意：我们不再直接连接 MySQL，而是通过 SSH 发送 docker exec 命令
# 这样用户不需要暴露 MySQL 端口，更安全
if not config:
print_warning("未配置服务器信息，跳过数据库更新")
return
try:
# 连接 SSH
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
print(f" 连接服务器: {config['host']}")
ssh.connect(
hostname=config['host'],
port=int(config.get('port', 22)),
username=config['username'],
password=config.get('password', ''),
timeout=30
)
print_success("SSH 连接成功")
# 构造 SQL 语句
# 更新 'default' 分组为最新版本 (更健壮，不依赖 ID=1)
sql = f"UPDATE plugin_groups SET current_version_file = 'core-v{version}.zip' WHERE name = 'default';"
# 构造 Docker 命令
# 假设容器名为 designercep_db，用户 designer_user，密码 DesignerPass123!，库名 designer_db
# 这些应该与 docker-compose.yml 保持一致
db_user = "designer_user"
db_pass = "DesignerPass123!"
db_name = "designer_db"
container_name = "designercep_db"
print(f" 执行远程 Docker SQL 命令...")
print(f" SQL: {sql}")
docker_cmd = f'docker exec -i {container_name} mysql -u{db_user} -p{db_pass} {db_name} -e "{sql}"'
stdin, stdout, stderr = ssh.exec_command(docker_cmd)
exit_status = stdout.channel.recv_exit_status()
if exit_status == 0:
print_success("数据库更新命令执行成功")
print(f" 输出: {stdout.read().decode().strip()}")
else:
print_error(f"数据库更新失败 (Exit code: {exit_status})")
print(f" 错误: {stderr.read().decode().strip()}")
raise Exception("Docker exec failed")
ssh.close()
except Exception as e:
print_error(f"数据库更新失败: {e}")
print_warning("您需要手动执行 SQL (进入容器执行):")
print(f" {sql}")
# 不抛出异常，以免打断流程，因为这步失败通常不影响文件上传
def step5_clean_cache():
"""步骤 5: 清除客户端缓存（测试用）"""
print_step(5, 8, "清除客户端缓存（可选）")
cache_dir = Path.home() / "AppData" / "Roaming" / "DesignerCache"
if not cache_dir.exists():
print_warning(f"缓存目录不存在: {cache_dir}")
return
try:
shutil.rmtree(cache_dir)
print_success(f"缓存已清除: {cache_dir}")
except Exception as e:
print_warning(f"清除缓存失败: {e}")
print(" 您可以手动删除缓存目录")
def step6_setup_config():
"""步骤 6: 配置服务器信息（首次运行）"""
print_step(6, 8, "配置服务器信息")
config = load_config()
if config:
print_warning("配置文件已存在，跳过配置")
print(f" 配置文件: {CONFIG_FILE}")
return config
print("请输入服务器配置信息：")
print()
config = {}
# SSH 配置
config['host'] = input(" 服务器地址: ").strip()
config['port'] = input(" SSH 端口 [22]: ").strip() or "22"
config['username'] = input(" SSH 用户名: ").strip()
config['password'] = input(" SSH 密码: ").strip()
config['remote_path'] = input(" 远程路径 [/var/www/DesignerCEP/Server/static]: ").strip() \
or "/var/www/DesignerCEP/Server/static"
print()
# MySQL 配置
use_mysql = input(" 是否配置 MySQL？ (y/n) [y]: ").strip().lower()
if use_mysql != 'n':
config['mysql'] = {}
config['mysql']['host'] = input(" MySQL 地址: ").strip()
config['mysql']['port'] = input(" MySQL 端口 [3306]: ").strip() or "3306"
config['mysql']['username'] = input(" MySQL 用户名: ").strip()
config['mysql']['password'] = input(" MySQL 密码: ").strip()
config['mysql']['database'] = input(" 数据库名: ").strip()
config['mysql']['table'] = input(" 表名 [plugin_groups]: ").strip() or "plugin_groups"
# 保存配置
save_config(config)
return config
def step7_summary(version, deployed):
"""步骤 7: 发布总结"""
print_step(7, 8, "发布总结")
print(f"""
{GREEN}{'='*60}
DesignerCEP 发布完成！
{'='*60}{RESET}
[发布信息]
版本号: {version}
构建时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
部署状态: {'已部署到服务器' if deployed else '仅本地构建'}
[构建产物]
Core: {DIST_CORE_DIR}
Shell: {DIST_SHELL_DIR}
Shell.zip: {DESIGNER_DIR / 'dist' / f'shell-{version}.zip'}
[后续步骤]
1. {'✓' if deployed else '○'} 验证服务器文件
2. {'✓' if deployed else '○'} 确认数据库版本
3. ○ 测试客户端更新功能
4. ○ 通知用户更新
[测试方法]
1. 删除客户端缓存（已自动执行）
2. 重启 Photoshop 插件
3. 登录账号，检查是否下载新版本
4. 验证功能是否正常
{GREEN}{'='*60}{RESET}
""")
# ==================== 主函数 ====================
def main():
parser = argparse.ArgumentParser(
description='DesignerCEP 自动发布脚本（Shell + Core）',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
# 仅构建（不部署）
python auto_deploy_core.py --version 1.0.6
# 构建并部署到服务器
python auto_deploy_core.py --version 1.0.6 --deploy
# 构建、部署并更新数据库
python auto_deploy_core.py --version 1.0.6 --deploy --update-db
# 首次运行，配置服务器信息
python auto_deploy_core.py --version 1.0.6 --setup
"""
)
parser.add_argument(
'--version', '-v',
required=True,
help='版本号，例如: 1.0.6（不要加 v）'
)
parser.add_argument(
'--deploy', '-d',
action='store_true',
help='部署到服务器'
)
parser.add_argument(
'--update-db',
action='store_true',
help='自动更新 MySQL 数据库'
)
parser.add_argument(
'--setup',
action='store_true',
help='配置服务器信息（首次运行）'
)
parser.add_argument(
'--skip-clean',
action='store_true',
help='跳过清除缓存步骤'
)
args = parser.parse_args()
version = args.version
print(f"""
{BLUE}{'='*60}
DesignerCEP 自动发布脚本
{'='*60}{RESET}
[配置信息]
版本号: {version}
项目根目录: {PROJECT_ROOT}
Designer: {DESIGNER_DIR}
部署到服务器: {'是' if args.deploy else '否'}
更新数据库: {'是' if args.update_db else '否'}
清除缓存: {'否' if args.skip_clean else '是'}
{BLUE}{'='*60}{RESET}
""")
try:
# 加载配置
config = None
if args.deploy or args.setup:
config = load_config()
if not config or args.setup:
config = step6_setup_config()
# 执行发布步骤
step1_build_frontend()
step2_package_shell_zip(version)
if args.deploy:
if not config:
print_error("未找到配置文件，请先运行 --setup 配置服务器")
sys.exit(1)
step3_upload_to_server(version, config)
if args.update_db:
step4_update_mysql(version, config)
else:
print_step(4, 8, "更新数据库（已跳过）")
print_warning("数据库未自动更新，请手动执行:")
print(f" UPDATE plugin_groups SET current_version = '{version}';")
else:
print_step(3, 8, "上传到服务器（已跳过）")
print_step(4, 8, "更新数据库（已跳过）")
if not args.skip_clean:
step5_clean_cache()
else:
print_step(5, 8, "清除缓存（已跳过）")
step7_summary(version, args.deploy)
except KeyboardInterrupt:
print_error("\n用户中断")
sys.exit(1)
except Exception as e:
print_error(f"发布失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()

461
AdminTool/config_interface.py Normal file
View File

@@ -0,0 +1,461 @@
import requests
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QHeaderView, QTableWidgetItem, QFrame,
QGridLayout, QLabel
)
from PyQt5.QtCore import Qt
from qfluentwidgets import (
Pivot, PushButton, TableWidget, CardWidget,
MessageBox, InfoBar, LineEdit, SpinBox, CheckBox, ComboBox,
FluentIcon as FIF, SubtitleLabel, CaptionLabel,
ScrollArea, BodyLabel, PrimaryPushButton, MessageBoxBase, StrongBodyLabel
)
class ConfigDialog(MessageBoxBase):
def __init__(self, title, parent=None):
super().__init__(parent)
self.titleLabel = SubtitleLabel(title, self)
self.viewLayout.addWidget(self.titleLabel)
self.widget.setMinimumWidth(350)
self.yesButton.setText("确定")
self.cancelButton.setText("取消")
def add_row(self, label_text, widget):
self.viewLayout.addWidget(StrongBodyLabel(label_text, self))
self.viewLayout.addWidget(widget)
class ConfigInterface(QWidget):
"""配置管理主界面"""
def __init__(self, api_client, parent=None):
super().__init__(parent)
self.api_client = api_client
# 主布局
self.main_layout = QVBoxLayout(self)
self.main_layout.setContentsMargins(30, 30, 30, 30)
self.main_layout.setSpacing(20)
# 标题
self.title_label = SubtitleLabel("系统配置", self)
self.main_layout.addWidget(self.title_label)
# Pivot 导航
self.pivot = Pivot(self)
self.main_layout.addWidget(self.pivot)
# 堆叠窗口容器
self.stacked_widget = QWidget()
self.stacked_layout = QVBoxLayout(self.stacked_widget)
self.stacked_layout.setContentsMargins(0, 0, 0, 0)
self.main_layout.addWidget(self.stacked_widget)
# 子页面
self.features_tab = FeaturesConfigTab(api_client, self)
self.vip_tab = VIPConfigTab(api_client, self)
self.checkin_tab = CheckInConfigTab(api_client, self)
self.stats_tab = StatsTab(api_client, self)
# 添加子页面到 Pivot
self.add_sub_interface(self.features_tab, 'features', '功能配置')
self.add_sub_interface(self.vip_tab, 'vip', 'VIP配置')
self.add_sub_interface(self.checkin_tab, 'checkin', '签到配置')
self.add_sub_interface(self.stats_tab, 'stats', '数据统计')
# 初始化显示第一个
self.pivot.setCurrentItem(self.features_tab.objectName())
self.pivot.currentItemChanged.connect(self.on_pivot_changed)
# 初始加载
self.features_tab.setVisible(True)
self.vip_tab.setVisible(False)
self.checkin_tab.setVisible(False)
self.stats_tab.setVisible(False)
# 默认布局填充
self.stacked_layout.addWidget(self.features_tab)
self.stacked_layout.addWidget(self.vip_tab)
self.stacked_layout.addWidget(self.checkin_tab)
self.stacked_layout.addWidget(self.stats_tab)
def add_sub_interface(self, widget, object_name, text):
widget.setObjectName(object_name)
self.pivot.addItem(routeKey=object_name, text=text)
def on_pivot_changed(self, route_key):
self.features_tab.setVisible(route_key == 'features')
self.vip_tab.setVisible(route_key == 'vip')
self.checkin_tab.setVisible(route_key == 'checkin')
self.stats_tab.setVisible(route_key == 'stats')
# 刷新数据
if route_key == 'features':
self.features_tab.load_data()
elif route_key == 'vip':
self.vip_tab.load_data()
elif route_key == 'checkin':
self.checkin_tab.load_data()
elif route_key == 'stats':
self.stats_tab.load_data()
class FeaturesConfigTab(QWidget):
"""功能配置标签页"""
def __init__(self, api_client, parent=None):
super().__init__(parent)
self.api_client = api_client
self.init_ui()
def init_ui(self):
self.v_layout = QVBoxLayout(self)
# 工具栏
self.tool_layout = QHBoxLayout()
self.btn_refresh = PushButton(FIF.SYNC, "刷新", self)
self.btn_refresh.clicked.connect(self.load_data)
self.btn_add = PrimaryPushButton(FIF.ADD, "新增功能", self)
self.btn_add.clicked.connect(self.add_feature)
self.tool_layout.addWidget(self.btn_refresh)
self.tool_layout.addWidget(self.btn_add)
self.tool_layout.addStretch(1)
self.v_layout.addLayout(self.tool_layout)
# 表格
self.table = TableWidget(self)
self.table.setColumnCount(7)
self.table.setHorizontalHeaderLabels([
"Key", "功能名称", "分类", "普通价格", "VIP价格", "SVIP价格", "状态"
])
self.table.verticalHeader().hide()
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.table.setSelectionBehavior(TableWidget.SelectRows)
self.table.setEditTriggers(TableWidget.NoEditTriggers)
self.table.doubleClicked.connect(self.edit_feature)
self.v_layout.addWidget(self.table)
def load_mock_data(self):
"""加载模拟数据"""
data = [
(1, 10, 0, 10),
(2, 10, 5, 15),
(3, 10, 10, 20),
(7, 10, 50, 60),
]
self.table.setRowCount(len(data))
for i, (days, base, bonus, total) in enumerate(data):
self.table.setItem(i, 0, QTableWidgetItem(f"{days} 天"))
self.table.setItem(i, 1, QTableWidgetItem(str(base)))
self.table.setItem(i, 2, QTableWidgetItem(str(bonus)))
self.table.setItem(i, 3, QTableWidgetItem(str(total)))
def load_data(self):
if not self.api_client:
InfoBar.warning("未连接", "请先连接到后端服务器", parent=self)
return
try:
resp = requests.get(f"{self.api_client.base_url}/admin/config/features", headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
data = resp.json()
self.table.setRowCount(len(data))
for i, item in enumerate(data):
self.table.setItem(i, 0, QTableWidgetItem(item.get('feature_key', '')))
self.table.setItem(i, 1, QTableWidgetItem(item.get('feature_name', '')))
self.table.setItem(i, 2, QTableWidgetItem(item.get('category', '')))
self.table.setItem(i, 3, QTableWidgetItem(str(item.get('points_cost', 0))))
self.table.setItem(i, 4, QTableWidgetItem(str(item.get('vip_points_cost', 0))))
self.table.setItem(i, 5, QTableWidgetItem(str(item.get('svip_points_cost', 0))))
enabled = item.get('enabled', False)
status_item = QTableWidgetItem("启用" if enabled else "禁用")
if enabled:
status_item.setForeground(Qt.darkGreen)
else:
status_item.setForeground(Qt.red)
self.table.setItem(i, 6, status_item)
# Store full data
self.table.item(i, 0).setData(Qt.UserRole, item)
InfoBar.success("加载成功", f"已加载 {len(data)} 个功能配置", parent=self)
except Exception as e:
InfoBar.error("加载失败", f"无法加载功能配置: {str(e)}", parent=self)
print(f"Error loading config: {e}")
def add_feature(self):
dialog = ConfigDialog("新增功能", self)
key_edit = LineEdit()
name_edit = LineEdit()
category_edit = LineEdit()
points_spin = SpinBox()
points_spin.setRange(0, 9999)
vip_spin = SpinBox()
vip_spin.setRange(0, 9999)
svip_spin = SpinBox()
svip_spin.setRange(0, 9999)
dialog.add_row("Key (唯一标识):", key_edit)
dialog.add_row("功能名称:", name_edit)
dialog.add_row("分类:", category_edit)
dialog.add_row("积分消耗:", points_spin)
dialog.add_row("VIP消耗:", vip_spin)
dialog.add_row("SVIP消耗:", svip_spin)
if dialog.exec():
key = key_edit.text().strip()
name = name_edit.text().strip()
if not key or not name:
InfoBar.warning("输入错误", "Key和名称不能为空", parent=self)
return
payload = {
"feature_key": key,
"feature_name": name,
"category": category_edit.text().strip(),
"points_cost": points_spin.value(),
"vip_points_cost": vip_spin.value(),
"svip_points_cost": svip_spin.value(),
"enabled": True
}
try:
resp = requests.post(f"{self.api_client.base_url}/admin/config/features", json=payload, headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
InfoBar.success("成功", f"功能 '{name}' 已添加", parent=self)
self.load_data()
except Exception as e:
InfoBar.error("失败", f"添加失败: {str(e)}", parent=self)
def edit_feature(self):
row = self.table.currentRow()
if row < 0: return
item = self.table.item(row, 0).data(Qt.UserRole)
dialog = ConfigDialog(f"编辑功能: {item['feature_name']}", self)
points_spin = SpinBox()
points_spin.setRange(0, 9999)
points_spin.setValue(item.get('points_cost', 0))
vip_spin = SpinBox()
vip_spin.setRange(0, 9999)
vip_spin.setValue(item.get('vip_points_cost', 0))
svip_spin = SpinBox()
svip_spin.setRange(0, 9999)
svip_spin.setValue(item.get('svip_points_cost', 0))
enabled_check = CheckBox("启用")
enabled_check.setChecked(item.get('enabled', True))
dialog.add_row("积分消耗:", points_spin)
dialog.add_row("VIP消耗:", vip_spin)
dialog.add_row("SVIP消耗:", svip_spin)
dialog.viewLayout.addWidget(enabled_check)
if dialog.exec():
payload = {
"points_cost": points_spin.value(),
"vip_points_cost": vip_spin.value(),
"svip_points_cost": svip_spin.value(),
"enabled": enabled_check.isChecked()
}
try:
resp = requests.put(f"{self.api_client.base_url}/admin/config/features/{item['feature_key']}", json=payload, headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
InfoBar.success("成功", "配置已更新", parent=self)
self.load_data()
except Exception as e:
InfoBar.error("失败", f"更新失败: {str(e)}", parent=self)
class VIPConfigTab(QWidget):
"""VIP配置标签页"""
def __init__(self, api_client, parent=None):
super().__init__(parent)
self.api_client = api_client
self.v_layout = QVBoxLayout(self)
self.v_layout.setContentsMargins(0, 0, 0, 0)
self.card = CardWidget(self)
self.card_layout = QGridLayout(self.card)
self.card_layout.setContentsMargins(20, 20, 20, 20)
self.card_layout.setSpacing(20)
self.price_spin = SpinBox()
self.price_spin.setRange(0, 9999)
self.price_spin.setValue(35)
self.card_layout.addWidget(StrongBodyLabel("VIP 价格 (元/月):", self.card), 0, 0)
self.card_layout.addWidget(self.price_spin, 0, 1)
self.quota_spin = SpinBox()
self.quota_spin.setRange(0, 9999)
self.quota_spin.setValue(25)
self.card_layout.addWidget(StrongBodyLabel("每日配额 (次):", self.card), 1, 0)
self.card_layout.addWidget(self.quota_spin, 1, 1)
self.multiplier_spin = SpinBox()
self.multiplier_spin.setRange(100, 1000)
self.multiplier_spin.setValue(160)
self.card_layout.addWidget(StrongBodyLabel("积分倍数 (%):", self.card), 2, 0)
self.card_layout.addWidget(self.multiplier_spin, 2, 1)
self.btn_save = PrimaryPushButton(FIF.SAVE, "保存配置", self.card)
self.btn_save.clicked.connect(self.save_data)
self.card_layout.addWidget(self.btn_save, 3, 0, 1, 2)
self.v_layout.addWidget(self.card)
self.v_layout.addStretch(1)
def load_data(self):
if not self.api_client:
return
try:
resp = requests.get(f"{self.api_client.base_url}/admin/config/vip", headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
data = resp.json()
# 查找VIP配置
for item in data:
if item.get('vip_type') == 'vip':
self.price_spin.setValue(int(item.get('price', 35)))
self.quota_spin.setValue(int(item.get('daily_quota', 25)))
self.multiplier_spin.setValue(int(item.get('points_multiplier', 1.6) * 100))
break
except Exception as e:
print(f"Load VIP config error: {e}")
def save_data(self):
if not self.api_client:
InfoBar.warning("未连接", "请先连接到后端服务器", parent=self)
return
try:
payload = {
"price": self.price_spin.value(),
"daily_quota": self.quota_spin.value(),
"points_multiplier": self.multiplier_spin.value() / 100.0
}
resp = requests.put(f"{self.api_client.base_url}/admin/config/vip/vip", json=payload, headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
InfoBar.success("保存成功", "VIP配置已更新", parent=self)
except Exception as e:
InfoBar.error("保存失败", str(e), parent=self)
class CheckInConfigTab(QWidget):
"""签到配置标签页"""
def __init__(self, api_client, parent=None):
super().__init__(parent)
self.api_client = api_client
self.v_layout = QVBoxLayout(self)
self.btn_add = PrimaryPushButton(FIF.ADD, "新增签到规则", self)
self.v_layout.addWidget(self.btn_add)
self.table = TableWidget(self)
self.table.setColumnCount(4)
self.table.setHorizontalHeaderLabels(["连续天数", "基础积分", "额外奖励", "总计"])
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.v_layout.addWidget(self.table)
self.load_mock_data()
def load_mock_data(self):
"""加载模拟数据"""
data = [
(1, 10, 0, 10),
(2, 10, 5, 15),
(3, 10, 10, 20),
(7, 10, 50, 60),
]
self.table.setRowCount(len(data))
for i, (days, base, bonus, total) in enumerate(data):
self.table.setItem(i, 0, QTableWidgetItem(f"{days} 天"))
self.table.setItem(i, 1, QTableWidgetItem(str(base)))
self.table.setItem(i, 2, QTableWidgetItem(str(bonus)))
self.table.setItem(i, 3, QTableWidgetItem(str(total)))
def load_data(self):
if not self.api_client:
return
try:
resp = requests.get(f"{self.api_client.base_url}/admin/config/checkin", headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
data = resp.json()
self.table.setRowCount(len(data))
for i, item in enumerate(data):
self.table.setItem(i, 0, QTableWidgetItem(f"{item.get('consecutive_days', 0)} 天"))
self.table.setItem(i, 1, QTableWidgetItem(str(item.get('base_points', 0))))
self.table.setItem(i, 2, QTableWidgetItem(str(item.get('bonus_points', 0))))
self.table.setItem(i, 3, QTableWidgetItem(str(item.get('total_points', 0))))
# Store full data
self.table.item(i, 0).setData(Qt.UserRole, item)
InfoBar.success("加载成功", f"已加载 {len(data)} 个签到规则", parent=self)
except Exception as e:
InfoBar.error("加载失败", str(e), parent=self)
class StatsTab(QWidget):
"""数据统计标签页"""
def __init__(self, api_client, parent=None):
super().__init__(parent)
self.api_client = api_client
self.v_layout = QVBoxLayout(self)
self.card = CardWidget(self)
self.card_layout = QVBoxLayout(self.card)
self.card_layout.addWidget(SubtitleLabel("今日数据概览", self.card))
self.grid = QGridLayout()
self.grid.addWidget(BodyLabel("今日活跃用户:"), 0, 0)
self.lbl_active_users = StrongBodyLabel("-", self.card)
self.grid.addWidget(self.lbl_active_users, 0, 1)
self.grid.addWidget(BodyLabel("今日签到:"), 1, 0)
self.lbl_checkins = StrongBodyLabel("-", self.card)
self.grid.addWidget(self.lbl_checkins, 1, 1)
self.grid.addWidget(BodyLabel("今日积分消耗:"), 2, 0)
self.lbl_points = StrongBodyLabel("-", self.card)
self.grid.addWidget(self.lbl_points, 2, 1)
self.grid.addWidget(BodyLabel("今日功能调用:"), 3, 0)
self.lbl_feature_usage = StrongBodyLabel("-", self.card)
self.grid.addWidget(self.lbl_feature_usage, 3, 1)
self.card_layout.addLayout(self.grid)
# 刷新按钮
self.btn_refresh = PrimaryPushButton(FIF.SYNC, "刷新统计", self.card)
self.btn_refresh.clicked.connect(self.load_data)
self.card_layout.addWidget(self.btn_refresh)
self.v_layout.addWidget(self.card)
self.v_layout.addStretch(1)
def load_data(self):
if not self.api_client:
InfoBar.warning("未连接", "请先连接到后端服务器", parent=self)
return
try:
resp = requests.get(f"{self.api_client.base_url}/admin/stats/today", headers=self.api_client.get_headers(), timeout=10)
resp.raise_for_status()
data = resp.json()
self.lbl_active_users.setText(str(data.get('active_users', 0)))
self.lbl_checkins.setText(str(data.get('check_ins', 0)))
self.lbl_points.setText(str(data.get('points_consumed', 0)))
self.lbl_feature_usage.setText(str(data.get('feature_usage', 0)))
InfoBar.success("刷新成功", "统计数据已更新", parent=self)
except Exception as e:
InfoBar.error("加载失败", f"无法加载统计数据: {str(e)}", parent=self)
print(f"Load stats error: {e}")

243
AdminTool/deploy_core_only.py
View File

@@ -1,243 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Core 快速发布脚本
仅执行：构建 Core -> 上传 Core -> 更新数据库
跳过 Shell 的构建和上传
"""
import os
import sys
import shutil
import subprocess
import argparse
import zipfile
import json
from pathlib import Path
from datetime import datetime
import paramiko
# ==================== 配置区域 ====================
PROJECT_ROOT = Path(__file__).parent.parent.absolute()
DESIGNER_DIR = PROJECT_ROOT / "Designer"
DIST_CORE_DIR = DESIGNER_DIR / "dist_core" # Core 构建输出目录
CONFIG_FILE = Path(__file__).parent / "deploy_config.json"
# 颜色输出
try:
import colorama
colorama.init()
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BLUE = '\033[94m'
RESET = '\033[0m'
except ImportError:
GREEN = YELLOW = RED = BLUE = RESET = ''
# ==================== 工具函数 ====================
def print_step(message):
print(f"\n{BLUE}{'='*60}{RESET}")
print(f"{GREEN}{message}{RESET}")
print(f"{BLUE}{'='*60}{RESET}\n")
def print_success(message):
print(f"{GREEN}✓ {message}{RESET}")
def print_error(message):
print(f"{RED}✗ {message}{RESET}")
def run_command(command, cwd=None):
print(f" 执行: {command}")
try:
result = subprocess.run(
command,
cwd=cwd,
shell=True,
check=True,
capture_output=True,
text=True,
encoding='utf-8',
errors='replace'
)
if result.stdout:
print(f" 输出: {result.stdout.strip()[:200]}...") # 只打印前200字符
return result
except subprocess.CalledProcessError as e:
print_error(f"命令执行失败: {e}")
if e.stderr:
print(f" 错误: {e.stderr}")
sys.exit(1)
def load_config():
if not CONFIG_FILE.exists():
print_error("未找到配置文件 deploy_config.json")
sys.exit(1)
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
# ==================== 核心步骤 ====================
def step1_build_core():
print_step("步骤 1: 构建 Core")
# 检查目录
if not DESIGNER_DIR.exists():
print_error(f"Designer 目录不存在: {DESIGNER_DIR}")
sys.exit(1)
# 执行构建
print(" 正在构建 Core (npm run build:core)...")
os.chdir(DESIGNER_DIR)
run_command("npm run build:core", cwd=DESIGNER_DIR)
# 验证
if not DIST_CORE_DIR.exists():
print_error(f"Core 构建失败，输出目录不存在: {DIST_CORE_DIR}")
sys.exit(1)
print_success("Core 构建完成")
def step2_upload_core(version, config):
print_step("步骤 2: 上传 Core 到服务器")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
print(f" 连接服务器: {config['host']}")
ssh.connect(
hostname=config['host'],
port=int(config.get('port', 22)),
username=config['username'],
password=config.get('password', ''),
timeout=30
)
print_success("SSH 连接成功")
sftp = ssh.open_sftp()
remote_base = config['remote_path']
remote_core_dir = f"{remote_base}/core/{version}"
# 创建远程目录
print(f" 创建远程目录: {remote_core_dir}")
ssh.exec_command(f"mkdir -p {remote_core_dir}")
# 递归上传
print(" 开始上传文件...")
upload_count = 0
for root, dirs, files in os.walk(DIST_CORE_DIR):
relative_root = Path(root).relative_to(DIST_CORE_DIR)
remote_root = f"{remote_core_dir}/{relative_root}".replace("\\", "/").rstrip("/")
if str(relative_root) == ".":
remote_root = remote_core_dir
# 确保子目录存在
try:
sftp.stat(remote_root)
except IOError:
sftp.mkdir(remote_root)
for file in files:
local_file = Path(root) / file
remote_file = f"{remote_root}/{file}"
sftp.put(str(local_file), remote_file)
upload_count += 1
if upload_count % 10 == 0:
print(f" 已上传 {upload_count} 个文件...", end='\r')
print(f"\n 共上传 {upload_count} 个文件")
print_success("Core 上传完成")
sftp.close()
ssh.close()
except Exception as e:
print_error(f"上传失败: {e}")
raise
def step3_update_db(version, config):
print_step("步骤 3: 更新数据库")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(
hostname=config['host'],
port=int(config.get('port', 22)),
username=config['username'],
password=config.get('password', ''),
timeout=30
)
# 构造 SQL
sql = f"UPDATE plugin_groups SET current_version_file = 'core-v{version}.zip' WHERE name = 'default';"
# 数据库配置
db_conf = config.get('mysql', {})
db_user = db_conf.get('username', 'designer_user')
db_pass = db_conf.get('password', 'DesignerPass123!')
db_name = db_conf.get('database', 'designer_db')
container_name = "designercep_db"
print(f" 更新 'default' 分组版本为: core-v{version}.zip")
docker_cmd = f'docker exec -i {container_name} mysql -u{db_user} -p{db_pass} {db_name} -e "{sql}"'
stdin, stdout, stderr = ssh.exec_command(docker_cmd)
exit_status = stdout.channel.recv_exit_status()
if exit_status == 0:
print_success("数据库更新成功")
else:
print_error(f"数据库更新失败: {stderr.read().decode().strip()}")
raise Exception("DB Update Failed")
ssh.close()
except Exception as e:
print_error(f"数据库操作失败: {e}")
raise
def step4_clean_cache():
print_step("步骤 4: 清除本地缓存")
cache_dir = Path.home() / "AppData" / "Roaming" / "DesignerCache"
if cache_dir.exists():
try:
shutil.rmtree(cache_dir)
print_success(f"已清除: {cache_dir}")
except Exception as e:
print_error(f"清除失败: {e}")
else:
print(" 无需清除")
# ==================== 主入口 ====================
def main():
parser = argparse.ArgumentParser(description='Core 快速发布脚本')
parser.add_argument('--version', '-v', required=True, help='版本号 (如 1.0.7)')
args = parser.parse_args()
version = args.version
print(f"{BLUE}开始发布 Core - 版本: {version}{RESET}")
try:
config = load_config()
step1_build_core()
step2_upload_core(version, config)
step3_update_db(version, config)
step4_clean_cache()
print_step("🎉 发布全部完成！")
print(f" 版本: {version}")
print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
except Exception as e:
print_error(f"\n发布过程中止: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

34
AdminTool/deploy_tool.py Normal file
View File

@@ -0,0 +1,34 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
快速启动部署工具（跳过组管理功能）
"""
import sys
from PyQt5.QtWidgets import QApplication
from admin_gui import AdminWindow
if __name__ == "__main__":
print("="*60)
print("🚀 启动部署工具...")
print("="*60)
print("\n💡 提示：如果看到后端 API 错误，可以忽略")
print(" 部署功能不依赖后端 API，可以正常使用\n")
app = QApplication(sys.argv)
window = AdminWindow()
# 直接切换到部署标签页
# FluentWindow uses switchTo
window.switchTo(window.deploy_interface)
# 启用标签页（即使连接失败）
# FluentWindow handles navigation enabling differently, but generally it's enabled by default
# window.tabs.setEnabled(True) # No longer needed/available
window.show()
print("✅ 工具已启动")
print(" 请切换到「自动化部署」标签页开始使用\n")
sys.exit(app.exec())

3
AdminTool/requirements.txt
View File

@@ -1,5 +1,4 @@
PyQt5
requests
paramiko
pymysql
colorama
PyQt-Fluent-Widgets

127
AdminTool/test_db_connection.py Normal file
View File

@@ -0,0 +1,127 @@
# -*- coding: utf-8 -*-
"""
测试数据库连接和 app_deployments 表（通过 SSH 隧道）
"""
import json
import pymysql
from datetime import datetime
import sys
import io
from sshtunnel import SSHTunnelForwarder
# 设置 stdout 为 UTF-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
def load_config():
with open('deploy_config.json', 'r', encoding='utf-8') as f:
return json.load(f)
def test_connection():
print("="*60)
print("测试 MySQL 数据库连接（通过 SSH 隧道）")
print("="*60)
config = load_config()
mysql_config = config['mysql']
print(f"\n📋 SSH 服务器:")
print(f" 地址: {config['host']}")
print(f" 端口: {config['port']}")
print(f" 用户: {config['username']}")
print(f"\n📋 MySQL 配置:")
print(f" 主机: {mysql_config['host']}")
print(f" 端口: {mysql_config['port']}")
print(f" 用户: {mysql_config['username']}")
print(f" 数据库: {mysql_config['database']}")
tunnel = None
try:
print("\n🔌 创建 SSH 隧道...")
tunnel = SSHTunnelForwarder(
(config['host'], int(config.get('port', 22))),
ssh_username=config['username'],
ssh_password=config['password'],
remote_bind_address=('127.0.0.1', int(mysql_config.get('port', 3306)))
)
tunnel.start()
print(f"✅ SSH 隧道已建立，本地端口: {tunnel.local_bind_port}")
print("\n🔌 连接 MySQL...")
conn = pymysql.connect(
host='127.0.0.1',
port=tunnel.local_bind_port,
user=mysql_config['username'],
password=mysql_config['password'],
database=mysql_config['database'],
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
print("✅ MySQL 连接成功！")
# 测试创建表
print("\n📝 创建/检查 app_deployments 表...")
with conn.cursor() as cursor:
sql = """
CREATE TABLE IF NOT EXISTS app_deployments (
id INT PRIMARY KEY AUTO_INCREMENT,
version VARCHAR(50) UNIQUE NOT NULL,
deployed_at DATETIME NOT NULL,
is_current BOOLEAN DEFAULT FALSE,
file_size_mb DECIMAL(10, 2),
comment VARCHAR(500),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
cursor.execute(sql)
conn.commit()
print("✅ 表已就绪")
# 查看表结构
print("\n📊 表结构:")
with conn.cursor() as cursor:
cursor.execute("DESCRIBE app_deployments")
for row in cursor.fetchall():
print(f" {row['Field']}: {row['Type']} {'(主键)' if row['Key'] == 'PRI' else ''}")
# 查看现有数据
print("\n📚 现有部署记录:")
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM app_deployments ORDER BY deployed_at DESC")
records = cursor.fetchall()
if records:
for i, rec in enumerate(records, 1):
status = "✅ 当前" if rec['is_current'] else ""
print(f"\n [{i}] {rec['version']} {status}")
print(f" 部署时间: {rec['deployed_at']}")
print(f" 大小: {rec['file_size_mb']} MB")
print(f" 备注: {rec['comment'] or '无'}")
else:
print(" (暂无记录)")
conn.close()
print("\n" + "="*60)
print("✅ 测试完成！数据库一切正常")
print("="*60)
except Exception as e:
print(f"\n❌ 测试失败: {e}")
print("\n请检查:")
print(" 1. SSH 服务器是否可访问")
print(" 2. deploy_config.json 配置是否正确")
print(" 3. 服务器上 MySQL 服务是否运行")
print(" 4. 数据库用户权限是否足够")
return False
finally:
if tunnel:
tunnel.stop()
print("\n🔌 SSH 隧道已关闭")
return True
if __name__ == "__main__":
test_connection()

24
AdminTool/test_launch.py Normal file
View File

@@ -0,0 +1,24 @@
import sys
from PyQt5.QtWidgets import QApplication
try:
from admin_gui import AdminWindow
print("Successfully imported AdminWindow")
except Exception as e:
print(f"Failed to import AdminWindow: {e}")
sys.exit(1)
def main():
app = QApplication(sys.argv)
try:
window = AdminWindow()
print("Successfully instantiated AdminWindow")
# Don't show or exec, just check if it crashes on init
# window.show()
except Exception as e:
print(f"Failed to instantiate AdminWindow: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()

113
AdminTool/test_version_system.py Normal file
View File

@@ -0,0 +1,113 @@
# -*- coding: utf-8 -*-
"""
测试版本管理系统（使用 SSH + JSON 文件）
"""
import json
import sys
import io
import paramiko
# 设置 stdout 为 UTF-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
def load_config():
with open('deploy_config.json', 'r', encoding='utf-8') as f:
return json.load(f)
def test_connection():
print("="*60)
print("测试版本管理系统（SSH + JSON 文件）")
print("="*60)
config = load_config()
print(f"\n📋 SSH 服务器:")
print(f" 地址: {config['host']}")
print(f" 端口: {config['port']}")
print(f" 用户: {config['username']}")
ssh = None
sftp = None
remote_file = '/var/www/app_versions/.deployments.json'
try:
print("\n🔌 连接 SSH...")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(
hostname=config['host'],
port=int(config.get('port', 22)),
username=config['username'],
password=config['password'],
timeout=10
)
print("✅ SSH 连接成功！")
# 测试创建目录
print("\n📁 检查/创建版本目录...")
stdin, stdout, stderr = ssh.exec_command('mkdir -p /var/www/app_versions')
stdout.channel.recv_exit_status()
print("✅ 目录已就绪: /var/www/app_versions/")
# 测试读写 JSON 文件
print(f"\n📝 测试版本记录文件: {remote_file}")
sftp = ssh.open_sftp()
try:
# 尝试读取现有文件
with sftp.open(remote_file, 'r') as f:
content = f.read().decode('utf-8')
data = json.loads(content)
print(f"✅ 文件已存在，读取成功")
except IOError:
# 文件不存在，创建新文件
print(" 文件不存在，创建新文件...")
data = {'deployments': []}
with sftp.open(remote_file, 'w') as f:
f.write(json.dumps(data, indent=2, ensure_ascii=False).encode('utf-8'))
print("✅ 文件创建成功")
# 显示现有记录
print("\n📚 现有部署记录:")
if data['deployments']:
for i, rec in enumerate(sorted(data['deployments'], key=lambda x: x['deployed_at'], reverse=True), 1):
status = "✅ 当前" if rec.get('is_current') else ""
print(f"\n [{i}] {rec['version']} {status}")
print(f" 部署时间: {rec['deployed_at']}")
print(f" 大小: {rec.get('file_size_mb', '-')} MB")
print(f" 备注: {rec.get('comment') or '无'}")
else:
print(" (暂无记录)")
# 测试写入
print("\n✏️ 测试写入...")
with sftp.open(remote_file, 'w') as f:
f.write(json.dumps(data, indent=2, ensure_ascii=False).encode('utf-8'))
print("✅ 写入成功")
print("\n" + "="*60)
print("✅ 测试完成！版本管理系统一切正常")
print("="*60)
print("\n💡 系统说明:")
print(" - 不需要 MySQL 数据库")
print(" - 版本信息存储在服务器的 JSON 文件中")
print(" - 文件位置: /var/www/app_versions/.deployments.json")
except Exception as e:
print(f"\n❌ 测试失败: {e}")
print("\n请检查:")
print(" 1. SSH 服务器是否可访问")
print(" 2. deploy_config.json 配置是否正确")
print(" 3. 服务器上是否有写入权限")
return False
finally:
if sftp:
sftp.close()
if ssh:
ssh.close()
return True
if __name__ == "__main__":
test_connection()