Files
feiqiu-ETL/etl_billiards/gui/widgets/env_editor.py
2026-01-27 22:47:05 +08:00

319 lines
11 KiB
Python

# -*- coding: utf-8 -*-
"""环境变量编辑器"""
from PySide6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGridLayout,
QGroupBox, QLabel, QLineEdit, QPushButton, QScrollArea,
QFrame, QMessageBox, QFileDialog, QCheckBox
)
from PySide6.QtCore import Qt, Signal
from ..utils.config_helper import ConfigHelper, ENV_GROUPS
class EnvEditor(QWidget):
"""环境变量编辑器"""
# 信号
config_saved = Signal() # 配置保存成功
def __init__(self, parent=None):
super().__init__(parent)
self.config_helper = ConfigHelper()
self.field_widgets = {} # 存储字段控件
self.show_sensitive = False
self._init_ui()
self.load_config()
def _init_ui(self):
"""初始化界面"""
layout = QVBoxLayout(self)
layout.setContentsMargins(16, 16, 16, 16)
layout.setSpacing(16)
# 标题和按钮
header_layout = QHBoxLayout()
title = QLabel("环境配置")
title.setProperty("heading", True)
header_layout.addWidget(title)
header_layout.addStretch()
self.show_sensitive_check = QCheckBox("显示敏感信息")
self.show_sensitive_check.stateChanged.connect(self._toggle_sensitive)
header_layout.addWidget(self.show_sensitive_check)
self.import_btn = QPushButton("导入")
self.import_btn.setProperty("secondary", True)
self.import_btn.clicked.connect(self._import_config)
header_layout.addWidget(self.import_btn)
self.export_btn = QPushButton("导出")
self.export_btn.setProperty("secondary", True)
self.export_btn.clicked.connect(self._export_config)
header_layout.addWidget(self.export_btn)
self.reload_btn = QPushButton("重新加载")
self.reload_btn.setProperty("secondary", True)
self.reload_btn.clicked.connect(self.load_config)
header_layout.addWidget(self.reload_btn)
self.save_btn = QPushButton("保存")
self.save_btn.clicked.connect(self._save_config)
header_layout.addWidget(self.save_btn)
layout.addLayout(header_layout)
# 配置文件路径
path_layout = QHBoxLayout()
path_layout.addWidget(QLabel("配置文件:"))
self.path_label = QLabel(str(self.config_helper.env_path))
self.path_label.setProperty("subheading", True)
path_layout.addWidget(self.path_label, 1)
layout.addLayout(path_layout)
# 滚动区域
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setFrameShape(QFrame.NoFrame)
layout.addWidget(scroll_area, 1)
# 配置组容器
config_widget = QWidget()
self.config_layout = QVBoxLayout(config_widget)
self.config_layout.setSpacing(16)
# 创建各配置组
self._create_config_groups()
# 弹性空间
self.config_layout.addStretch()
scroll_area.setWidget(config_widget)
# 验证结果
self.validation_label = QLabel()
self.validation_label.setWordWrap(True)
layout.addWidget(self.validation_label)
def _create_config_groups(self):
"""创建配置分组"""
for group_id, group_info in ENV_GROUPS.items():
group = QGroupBox(group_info["title"])
grid_layout = QGridLayout(group)
for row, key in enumerate(group_info["keys"]):
# 标签
label = QLabel(f"{key}:")
label.setMinimumWidth(180)
grid_layout.addWidget(label, row, 0)
# 输入框
edit = QLineEdit()
edit.setPlaceholderText(self._get_placeholder(key))
# 敏感字段处理
if key in group_info.get("sensitive", []):
edit.setEchoMode(QLineEdit.Password)
edit.setProperty("sensitive", True)
edit.textChanged.connect(self._on_value_changed)
grid_layout.addWidget(edit, row, 1)
# 存储控件引用
self.field_widgets[key] = edit
self.config_layout.addWidget(group)
# 其他配置组(动态添加)
self.other_group = QGroupBox("其他配置")
self.other_layout = QGridLayout(self.other_group)
self.other_group.setVisible(False)
self.config_layout.addWidget(self.other_group)
def load_config(self):
"""加载配置"""
env_vars = self.config_helper.load_env()
# 更新已知字段
for key, edit in self.field_widgets.items():
value = env_vars.get(key, "")
edit.blockSignals(True)
edit.setText(value)
edit.blockSignals(False)
# 处理其他字段
known_keys = set(self.field_widgets.keys())
other_keys = [k for k in env_vars.keys() if k not in known_keys]
# 清除旧的其他字段
while self.other_layout.count():
item = self.other_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
# 添加其他字段
if other_keys:
self.other_group.setVisible(True)
for row, key in enumerate(sorted(other_keys)):
label = QLabel(f"{key}:")
self.other_layout.addWidget(label, row, 0)
edit = QLineEdit(env_vars[key])
edit.textChanged.connect(self._on_value_changed)
self.other_layout.addWidget(edit, row, 1)
self.field_widgets[key] = edit
else:
self.other_group.setVisible(False)
self._validate()
def _save_config(self):
"""保存配置"""
# 收集所有值
env_vars = {}
for key, edit in self.field_widgets.items():
value = edit.text().strip()
if value:
env_vars[key] = value
# 验证
errors = self.config_helper.validate_env(env_vars)
if errors:
reply = QMessageBox.question(
self,
"验证警告",
"配置存在以下问题:\n\n" + "\n".join(f"{e}" for e in errors) + "\n\n是否仍要保存?",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No
)
if reply == QMessageBox.No:
return
# 保存
if self.config_helper.save_env(env_vars):
QMessageBox.information(self, "成功", "配置已保存")
self.config_saved.emit()
else:
QMessageBox.critical(self, "错误", "保存配置失败")
def _import_config(self):
"""导入配置"""
file_path, _ = QFileDialog.getOpenFileName(
self,
"导入配置文件",
"",
"环境文件 (*.env);;所有文件 (*.*)"
)
if not file_path:
return
try:
from pathlib import Path
temp_helper = ConfigHelper(Path(file_path))
env_vars = temp_helper.load_env()
# 更新字段
for key, value in env_vars.items():
if key in self.field_widgets:
self.field_widgets[key].setText(value)
QMessageBox.information(self, "成功", f"已导入 {len(env_vars)} 个配置项")
except Exception as e:
QMessageBox.critical(self, "错误", f"导入失败: {e}")
def _export_config(self):
"""导出配置"""
file_path, _ = QFileDialog.getSaveFileName(
self,
"导出配置文件",
".env.backup",
"环境文件 (*.env);;所有文件 (*.*)"
)
if not file_path:
return
try:
from pathlib import Path
# 收集当前值
env_vars = {}
for key, edit in self.field_widgets.items():
value = edit.text().strip()
if value:
env_vars[key] = value
# 保存到指定路径
temp_helper = ConfigHelper(Path(file_path))
if temp_helper.save_env(env_vars):
QMessageBox.information(self, "成功", f"配置已导出到:\n{file_path}")
else:
QMessageBox.critical(self, "错误", "导出失败")
except Exception as e:
QMessageBox.critical(self, "错误", f"导出失败: {e}")
def _toggle_sensitive(self, state: int):
"""切换敏感信息显示"""
self.show_sensitive = state == Qt.Checked
for key, edit in self.field_widgets.items():
if edit.property("sensitive"):
edit.setEchoMode(
QLineEdit.Normal if self.show_sensitive else QLineEdit.Password
)
def _on_value_changed(self):
"""值变化时验证"""
self._validate()
def _validate(self):
"""验证配置"""
env_vars = {}
for key, edit in self.field_widgets.items():
value = edit.text().strip()
if value:
env_vars[key] = value
errors = self.config_helper.validate_env(env_vars)
if errors:
self.validation_label.setText("" + "; ".join(errors))
self.validation_label.setProperty("status", "warning")
else:
self.validation_label.setText("✓ 配置验证通过")
self.validation_label.setProperty("status", "success")
self.validation_label.style().unpolish(self.validation_label)
self.validation_label.style().polish(self.validation_label)
@staticmethod
def _get_placeholder(key: str) -> str:
"""获取占位符提示"""
placeholders = {
"PG_DSN": "postgresql://user:password@host:5432/dbname",
"PG_HOST": "localhost",
"PG_PORT": "5432",
"PG_NAME": "billiards",
"PG_USER": "postgres",
"PG_PASSWORD": "密码",
"API_BASE": "https://pc.ficoo.vip/apiprod/admin/v1",
"API_TOKEN": "Bearer token",
"API_TIMEOUT": "20",
"API_PAGE_SIZE": "200",
"STORE_ID": "门店ID (数字)",
"TIMEZONE": "Asia/Taipei",
"EXPORT_ROOT": "export/JSON",
"LOG_ROOT": "export/LOG",
"FETCH_ROOT": "JSON 抓取输出目录",
"INGEST_SOURCE_DIR": "本地 JSON 输入目录",
"PIPELINE_FLOW": "FULL / FETCH_ONLY / INGEST_ONLY",
"RUN_TASKS": "任务列表,逗号分隔",
"OVERLAP_SECONDS": "3600",
"WINDOW_START": "2025-07-01 00:00:00",
"WINDOW_END": "2025-08-01 00:00:00",
}
return placeholders.get(key, "")