BigMeng/jiateng_ws
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

多数据源配置

Browse Source
This commit is contained in:
zhu-mengmeng 2025-06-13 17:14:03 +08:00
parent 935fd44f78
commit e9f94db1d6
11 changed files with 505 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
config/app_config.json
View File

@ -9,13 +9,29 @@
}
},
"database": {
"type": "sqlite",
"default": "sqlite",
"sources": {
"sqlite": {
"path": "db/jtDB.db",
"host": "",
"port": "",
"user": "",
"description": "默认SQLite数据库"
},
"postgresql": {
"host": "localhost",
"port": "5432",
"user": "postgres",
"password": "",
"name": ""
"name": "jtDB",
"description": "PostgreSQL数据库"
},
"mysql": {
"host": "localhost",
"port": "3306",
"user": "root",
"password": "",
"name": "jtDB",
"description": "MySQL数据库"
}
}
},
"camera": {
"enabled": false,

12
main.py
View File

@ -179,8 +179,13 @@ def main():
# 创建db目录（如果不存在）
os.makedirs('db', exist_ok=True)
# 从配置获取SQLite数据库路径
config_loader = ConfigLoader.get_instance()
sqlite_config = config_loader.get_database_config('sqlite')
sqlite_db_path = sqlite_config.get('path', 'db/jtDB.db')
# 检查数据库是否存在，如果不存在则初始化
if not os.path.exists('db/jtDB.db'):
if not os.path.exists(sqlite_db_path):
from utils.init_db import init_database
init_database()
logging.info("初始化数据库完成")
@ -190,6 +195,11 @@ def main():
exit_code = app.exec()
logging.info(f"应用程序退出，退出码: {exit_code}")
# 关闭所有数据库连接
from utils.sql_utils import SQLUtils
SQLUtils.close_all_connections()
sys.exit(exit_code)
except Exception as e:

34
ui/settings_ui.py
View File

@ -205,25 +205,31 @@ class SettingsUI(QWidget):
self.database_layout = QVBoxLayout(self.database_tab)
self.database_layout.setContentsMargins(20, 20, 20, 20)
# 数据库类型选择
self.db_type_group = QGroupBox("数据库类型")
# 数据源类型选择
self.db_type_group = QGroupBox("数据源类型")
self.db_type_group.setFont(self.normal_font)
self.db_type_layout = QHBoxLayout()
self.sqlite_radio = QCheckBox("SQLite")
self.sqlite_radio.setFont(self.normal_font)
self.sqlite_radio.setChecked(True)
self.db_type_combo = QComboBox()
self.db_type_combo.setFont(self.normal_font)
self.db_type_combo.addItem("SQLite")
self.db_type_combo.addItem("PostgreSQL")
self.db_type_combo.addItem("MySQL")
self.pgsql_radio = QCheckBox("PostgreSQL")
self.pgsql_radio.setFont(self.normal_font)
self.mysql_radio = QCheckBox("MySQL")
self.mysql_radio.setFont(self.normal_font)
self.db_type_layout.addWidget(self.sqlite_radio)
self.db_type_layout.addWidget(self.pgsql_radio)
self.db_type_layout.addWidget(self.mysql_radio)
self.db_type_layout.addWidget(QLabel("当前配置类型："))
self.db_type_layout.addWidget(self.db_type_combo)
self.db_type_layout.addStretch(1)
# 添加当前使用的数据源选择
self.current_source_label = QLabel("当前使用的数据源：")
self.current_source_label.setFont(self.normal_font)
self.current_source_combo = QComboBox()
self.current_source_combo.setFont(self.normal_font)
self.current_source_combo.addItems(["SQLite", "PostgreSQL", "MySQL"])
self.db_type_layout.addWidget(self.current_source_label)
self.db_type_layout.addWidget(self.current_source_combo)
self.db_type_group.setLayout(self.db_type_layout)
self.database_layout.addWidget(self.db_type_group)

BIN
utils/__pycache__/config_loader.cpython-310.pyc
View File

Binary file not shown.

BIN
utils/__pycache__/sql_utils.cpython-310.pyc
View File

Binary file not shown.

147
utils/config_loader.py
View File

@ -32,6 +32,9 @@ class ConfigLoader:
with open(self.config_file, 'r', encoding='utf-8') as f:
self.config = json.load(f)
logging.info(f"已加载配置文件: {self.config_file}")
# 检查并升级配置文件结构（兼容旧版本的配置）
self._upgrade_config_if_needed()
else:
# 创建默认配置
self.config = {
@ -40,17 +43,50 @@ class ConfigLoader:
"version": "1.0.0",
"features": {
"enable_serial_ports": False,
"enable_keyboard_listener": False
"enable_keyboard_listener": False,
"enable_camera": False
}
},
"database": {
"type": "sqlite",
"current": "sqlite",
"sources": {
"sqlite": {
"path": "db/jtDB.db",
"host": "",
"port": "",
"user": "",
"description": "默认SQLite数据库"
},
"postgresql": {
"host": "localhost",
"port": "5432",
"user": "postgres",
"password": "",
"name": ""
"name": "jtDB",
"description": ""
},
"mysql": {
"host": "localhost",
"port": "3306",
"user": "root",
"password": "",
"name": "jtDB",
"description": ""
}
}
},
"camera": {
"enabled": False,
"default_exposure": 20000,
"default_gain": 10,
"default_framerate": 30
},
"modbus": {
"host": "localhost",
"port": "5020"
},
"serial": {
"keyboard": {
"trigger_key": "Key.page_up",
"enabled": False
}
}
}
@ -66,10 +102,86 @@ class ConfigLoader:
"version": "1.0.0",
"features": {
"enable_serial_ports": False,
"enable_keyboard_listener": False
"enable_keyboard_listener": False,
"enable_camera": False
}
},
"database": {
"current": "sqlite",
"sources": {
"sqlite": {
"path": "db/jtDB.db",
"description": "默认SQLite数据库"
}
}
}
}
def _upgrade_config_if_needed(self):
"""升级配置文件结构（兼容旧版本的配置）"""
try:
# 检查数据库配置是否需要升级
if 'database' in self.config:
db_config = self.config['database']
# 旧版本配置结构检查
if 'type' in db_config and 'sources' not in db_config:
logging.info("检测到旧版本的数据库配置，正在升级...")
# 获取旧配置
db_type = db_config.get('type', 'sqlite')
# 创建新的配置结构
new_db_config = {
"default": db_type,
"sources": {}
}
# 转移旧配置到新结构
if db_type == 'sqlite':
new_db_config['sources']['sqlite'] = {
"path": db_config.get('path', 'db/jtDB.db'),
"description": "从旧版本升级的SQLite数据库"
}
else:
new_db_config['sources'][db_type] = {
"host": db_config.get('host', 'localhost'),
"port": db_config.get('port', '5432' if db_type == 'postgresql' else '3306'),
"user": db_config.get('user', ''),
"password": db_config.get('password', ''),
"name": db_config.get('name', 'jtDB'),
"description": f"从旧版本升级的{db_type}数据库"
}
# 确保至少有一个sqlite配置
if 'sqlite' not in new_db_config['sources']:
new_db_config['sources']['sqlite'] = {
"path": "db/jtDB.db",
"description": "默认SQLite数据库"
}
# 更新配置
self.config['database'] = new_db_config
self.save_config()
logging.info("数据库配置已成功升级")
# 将current改为default
if 'current' in db_config and 'default' not in db_config:
logging.info("将数据库配置中的'current'改为'default'...")
db_config['default'] = db_config.pop('current')
self.save_config()
# 将pgsql改为postgresql
if 'sources' in db_config and 'pgsql' in db_config['sources'] and 'postgresql' not in db_config['sources']:
logging.info("将数据库配置中的'pgsql'改为'postgresql'...")
db_config['sources']['postgresql'] = db_config['sources'].pop('pgsql')
# 如果默认值是pgsql，也更新
if db_config.get('default') == 'pgsql':
db_config['default'] = 'postgresql'
self.save_config()
except Exception as e:
logging.error(f"升级配置文件结构失败: {e}")
def save_config(self):
"""保存配置到文件"""
@ -164,3 +276,24 @@ class ConfigLoader:
self.config['serial'][key] = config_data
# 这里不保存配置，等待调用save_config方法时一并保存
return True
def get_database_config(self, db_type=None):
"""
获取指定类型的数据库配置
Args:
db_type: 数据库类型，如'sqlite', 'postgresql', 'mysql'等，不指定则使用默认配置
Returns:
dict: 数据库配置
"""
# 如果未指定数据库类型，使用当前设置的类型
if db_type is None:
db_type = self.get_value('database.default', 'sqlite')
# 处理pgsql和postgresql兼容
if db_type == 'pgsql':
db_type = 'postgresql'
# 获取数据库配置
return self.get_value(f'database.sources.{db_type}', {})

13
utils/init_db.py
View File

@ -2,9 +2,20 @@ from utils.sql_utils import SQLUtils
import datetime
import os
import logging
from utils.config_loader import ConfigLoader
def init_database():
db = SQLUtils('sqlite', database='db/jtDB.db')
# 获取SQLite数据源的路径
config_loader = ConfigLoader.get_instance()
sqlite_config = config_loader.get_database_config('sqlite')
db_path = sqlite_config.get('path', 'db/jtDB.db')
# 确保db目录存在
db_dir = os.path.dirname(db_path)
os.makedirs(db_dir, exist_ok=True)
logging.info(f"初始化数据库: {db_path}")
db = SQLUtils('sqlite', database=db_path)
try:
db.begin_transaction()

149
utils/sql_utils.py
View File

@ -1,4 +1,6 @@
import sys
import logging
from utils.config_loader import ConfigLoader
try:
import psycopg2
@ -10,27 +12,127 @@ try:
except ImportError:
sqlite3 = None
try:
import mysql.connector
except ImportError:
mysql = None
class SQLUtils:
def __init__(self, db_type, **kwargs):
self.db_type = db_type.lower()
# 存储连接池，避免重复创建连接
_connection_pool = {}
def __init__(self, db_type=None, source_name=None, **kwargs):
"""初始化SQLUtils对象
Args:
db_type: 数据库类型 'sqlite', 'postgresql', 'mysql' 之一，如果为None则使用配置文件中的默认数据源
source_name: 数据源名称，用于从配置中获取特定的数据源，如'sqlite', 'postgresql', 'mysql'
**kwargs: 连接参数，如果没有提供，则使用配置文件中的参数
"""
self.conn = None
self.cursor = None
# 如果指定了source_name，直接使用该名称的数据源配置
if source_name:
config_loader = ConfigLoader.get_instance()
db_config = config_loader.get_value(f'database.sources.{source_name}', {})
if not db_config:
raise ValueError(f"未找到数据源配置: {source_name}")
db_type = source_name # 数据源名称同时也是类型
if not kwargs: # 如果没有提供连接参数，则使用配置中的参数
if source_name == 'sqlite':
kwargs = {'database': db_config.get('path', 'db/jtDB.db')}
else:
kwargs = {
'host': db_config.get('host', 'localhost'),
'user': db_config.get('user', ''),
'password': db_config.get('password', ''),
'database': db_config.get('name', 'jtDB')
}
if 'port' in db_config and db_config['port']:
kwargs['port'] = int(db_config['port'])
# 如果没有指定数据库类型和数据源名称，则使用配置中的默认数据源
elif db_type is None:
config_loader = ConfigLoader.get_instance()
default_source = config_loader.get_value('database.default', 'sqlite')
# 如果没有提供连接参数，则从配置文件获取
if not kwargs:
db_config = config_loader.get_database_config(default_source)
if default_source == 'sqlite':
kwargs = {'database': db_config.get('path', 'db/jtDB.db')}
else:
kwargs = {
'host': db_config.get('host', 'localhost'),
'user': db_config.get('user', ''),
'password': db_config.get('password', ''),
'database': db_config.get('name', 'jtDB')
}
if 'port' in db_config and db_config['port']:
kwargs['port'] = int(db_config['port'])
db_type = default_source
self.db_type = db_type.lower()
self.kwargs = kwargs
self.source_name = source_name or self.db_type
# 尝试从连接池获取连接，如果没有则创建新连接
self._get_connection()
def _get_connection(self):
"""从连接池获取连接，如果没有则创建新连接"""
# 创建连接键，包含数据库类型和连接参数
conn_key = f"{self.db_type}:{str(self.kwargs)}"
# 检查连接池中是否已有此连接
if conn_key in SQLUtils._connection_pool:
try:
# 尝试执行简单查询，确认连接有效
conn, cursor = SQLUtils._connection_pool[conn_key]
cursor.execute("SELECT 1")
# 连接有效，直接使用
self.conn = conn
self.cursor = cursor
return
except Exception:
# 连接已失效，从连接池移除
del SQLUtils._connection_pool[conn_key]
# 创建新连接
self.connect()
# 将新连接添加到连接池
if self.conn and self.cursor:
SQLUtils._connection_pool[conn_key] = (self.conn, self.cursor)
def connect(self):
if self.db_type == 'pgsql' or self.db_type == 'postgresql':
"""连接到数据库"""
try:
if self.db_type in ['pgsql', 'postgresql']:
if not psycopg2:
raise ImportError('psycopg2 is not installed')
self.conn = psycopg2.connect(**self.kwargs)
elif self.db_type == 'sqlite' or self.db_type == 'sqlite3':
elif self.db_type in ['sqlite', 'sqlite3']:
if not sqlite3:
raise ImportError('sqlite3 is not installed')
self.conn = sqlite3.connect(self.kwargs.get('database', ':memory:'))
elif self.db_type == 'mysql':
if not mysql:
raise ImportError('mysql.connector is not installed')
self.conn = mysql.connector.connect(**self.kwargs)
else:
raise ValueError(f'Unsupported db_type: {self.db_type}')
raise ValueError(f'不支持的数据库类型: {self.db_type}')
self.cursor = self.conn.cursor()
logging.debug(f"成功连接到数据库: {self.db_type}")
except Exception as e:
logging.error(f"连接数据库失败: {e}")
raise
def execute_query(self, sql, params=None):
if params is None:
@ -72,7 +174,36 @@ class SQLUtils:
return self.cursor.fetchall()
def close(self):
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()
"""关闭连接（实际上是将连接返回到连接池）"""
# 这里不再实际关闭连接，让连接池管理连接生命周期
pass
@staticmethod
def close_all_connections():
"""关闭所有连接池中的连接"""
for conn, cursor in SQLUtils._connection_pool.values():
try:
if cursor:
cursor.close()
if conn:
conn.close()
except Exception as e:
logging.error(f"关闭数据库连接失败: {e}")
SQLUtils._connection_pool.clear()
logging.info("已关闭所有数据库连接")
@staticmethod
def get_sqlite_connection():
"""获取SQLite连接"""
return SQLUtils(source_name='sqlite')
@staticmethod
def get_postgresql_connection():
"""获取PostgreSQL连接"""
return SQLUtils(source_name='postgresql')
@staticmethod
def get_mysql_connection():
"""获取MySQL连接"""
return SQLUtils(source_name='mysql')

BIN
widgets/__pycache__/login_widget.cpython-310.pyc
View File

Binary file not shown.

6
widgets/login_widget.py
View File

@ -10,7 +10,8 @@ import threading
def check_user_login(user_id, password):
"""验证用户登录"""
try:
db = SQLUtils('sqlite', database='db/jtDB.db')
# 始终使用SQLite数据源验证登录
db = SQLUtils(source_name='sqlite')
db.execute_query("SELECT id FROM user WHERE username = ? AND password = ? AND is_deleted = 0", (user_id, password))
result = db.fetchone()
db.close()
@ -22,7 +23,8 @@ def check_user_login(user_id, password):
def get_user_info(user_id):
"""获取用户信息"""
try:
db = SQLUtils('sqlite', database='db/jtDB.db')
# 始终使用SQLite数据源获取用户信息
db = SQLUtils(source_name='sqlite')
db.execute_query("SELECT username, 'Default Corp', 1, 1 FROM user WHERE username = ?", (user_id,))
result = db.fetchone()
db.close()

162
widgets/settings_widget.py
View File

@ -1,9 +1,12 @@
from PySide6.QtWidgets import QMessageBox, QVBoxLayout
import logging
import json
import os
from ui.settings_ui import SettingsUI
from utils.sql_utils import SQLUtils
from widgets.inspection_settings_widget import InspectionSettingsWidget
from widgets.pallet_type_settings_widget import PalletTypeSettingsWidget
from utils.config_loader import ConfigLoader
class SettingsWidget(SettingsUI):
def __init__(self, parent=None):
@ -52,19 +55,23 @@ class SettingsWidget(SettingsUI):
else:
logging.error("无法找到pallet_type_layout布局")
# 加载配置文件
self.config_loader = ConfigLoader.get_instance()
# 连接信号和槽
self.connect_signals()
# 初始化数据库类型UI状态
self.update_db_ui_state()
self.load_db_config()
logging.info("SettingsWidget初始化完成")
def connect_signals(self):
# 数据库类型选择
self.sqlite_radio.toggled.connect(self.update_db_ui_state)
self.pgsql_radio.toggled.connect(self.update_db_ui_state)
self.mysql_radio.toggled.connect(self.update_db_ui_state)
self.db_type_combo.currentTextChanged.connect(self.update_db_ui_state)
# 更新当前使用的数据源
self.current_source_combo.currentTextChanged.connect(self.update_default_source)
# 按钮动作
self.test_conn_button.clicked.connect(self.test_connection)
@ -80,10 +87,67 @@ class SettingsWidget(SettingsUI):
if hasattr(self, 'back_button'):
self.back_button.clicked.connect(self.back_to_main)
def load_db_config(self):
"""加载数据库配置"""
try:
# 获取默认数据源
default_source = self.config_loader.get_value('database.default', 'sqlite').lower()
# 设置当前使用的数据源组合框
self._update_source_combo_items() # 更新组合框项目
index = self.current_source_combo.findText(default_source.capitalize(), Qt.MatchFixedString)
if index >= 0:
self.current_source_combo.setCurrentIndex(index)
# 默认选择当前使用的数据源类型
index = self.db_type_combo.findText(default_source.capitalize(), Qt.MatchFixedString)
if index >= 0:
self.db_type_combo.setCurrentIndex(index)
# 更新UI状态
self.update_db_ui_state()
except Exception as e:
logging.error(f"加载数据库配置失败: {e}")
def _update_source_combo_items(self):
"""更新数据源下拉框项目"""
try:
# 清空当前项目
self.current_source_combo.clear()
# 获取所有配置的数据源
sources = self.config_loader.get_value('database.sources', {})
# 添加数据源到下拉框
for source_name in sources.keys():
# 处理特殊情况：postgresql显示为PostgreSQL
display_name = source_name.capitalize()
self.current_source_combo.addItem(display_name, source_name)
except Exception as e:
logging.error(f"更新数据源下拉框失败: {e}")
# 添加默认项
self.current_source_combo.addItem("SQLite", "sqlite")
def update_db_ui_state(self):
"""根据选择的数据库类型更新UI状态"""
if self.sqlite_radio.isChecked():
# SQLite模式下，只需要数据库文件路径
db_type = self.db_type_combo.currentText().lower()
# 处理特殊情况：PostgreSQL对应postgresql
if db_type == "postgresql":
db_type = "postgresql"
# 加载选定类型的数据源配置
config_path = f"database.sources.{db_type}"
db_config = {}
if db_type == "sqlite":
# SQLite模式下，只需要数据库文件路径和说明
path = self.config_loader.get_value(f"{config_path}.path", "db/jtDB.db")
description = self.config_loader.get_value(f"{config_path}.description", "")
self.host_input.setEnabled(False)
self.host_input.setText("")
self.user_input.setEnabled(False)
@ -93,41 +157,60 @@ class SettingsWidget(SettingsUI):
self.port_input.setEnabled(False)
self.port_input.setText("")
self.database_input.setEnabled(True)
self.database_input.setText("db/jtDB.db")
elif self.pgsql_radio.isChecked():
self.database_input.setText(path)
self.desc_input.setText(description)
elif db_type == "postgresql":
# PostgreSQL模式下，需要完整的连接信息
host = self.config_loader.get_value(f"{config_path}.host", "localhost")
port = self.config_loader.get_value(f"{config_path}.port", "5432")
user = self.config_loader.get_value(f"{config_path}.user", "postgres")
password = self.config_loader.get_value(f"{config_path}.password", "")
name = self.config_loader.get_value(f"{config_path}.name", "jtDB")
description = self.config_loader.get_value(f"{config_path}.description", "")
self.host_input.setEnabled(True)
self.host_input.setText("localhost")
self.host_input.setText(host)
self.user_input.setEnabled(True)
self.user_input.setText("postgres")
self.user_input.setText(user)
self.password_input.setEnabled(True)
self.password_input.setText("")
self.password_input.setText(password)
self.port_input.setEnabled(True)
self.port_input.setText("5432")
self.port_input.setText(port)
self.database_input.setEnabled(True)
self.database_input.setText("jtDB")
elif self.mysql_radio.isChecked():
self.database_input.setText(name)
self.desc_input.setText(description)
elif db_type == "mysql":
# MySQL模式下，需要完整的连接信息
host = self.config_loader.get_value(f"{config_path}.host", "localhost")
port = self.config_loader.get_value(f"{config_path}.port", "3306")
user = self.config_loader.get_value(f"{config_path}.user", "root")
password = self.config_loader.get_value(f"{config_path}.password", "")
name = self.config_loader.get_value(f"{config_path}.name", "jtDB")
description = self.config_loader.get_value(f"{config_path}.description", "")
self.host_input.setEnabled(True)
self.host_input.setText("localhost")
self.host_input.setText(host)
self.user_input.setEnabled(True)
self.user_input.setText("root")
self.user_input.setText(user)
self.password_input.setEnabled(True)
self.password_input.setText("")
self.password_input.setText(password)
self.port_input.setEnabled(True)
self.port_input.setText("3306")
self.port_input.setText(port)
self.database_input.setEnabled(True)
self.database_input.setText("jtDB")
self.database_input.setText(name)
self.desc_input.setText(description)
def update_default_source(self):
"""更新默认使用的数据源"""
default_source = self.current_source_combo.currentText().lower()
self.config_loader.set_value('database.default', default_source)
logging.info(f"已更新默认使用的数据源为: {default_source}")
def get_db_type(self):
"""获取当前选择的数据库类型"""
if self.sqlite_radio.isChecked():
return "sqlite"
elif self.pgsql_radio.isChecked():
return "pgsql"
elif self.mysql_radio.isChecked():
return "mysql"
return "sqlite" # 默认返回sqlite
return self.db_type_combo.currentText().lower()
def get_connection_params(self):
"""获取数据库连接参数"""
@ -174,8 +257,25 @@ class SettingsWidget(SettingsUI):
params = self.get_connection_params()[1]
desc = self.desc_input.text().strip()
# 这里应该将设置保存到配置文件中
# 为了简单起见，这里只显示一个消息框
try:
# 构建要保存的配置数据
config_path = f"database.sources.{db_type}"
if db_type == "sqlite":
self.config_loader.set_value(f"{config_path}.path", params["database"])
self.config_loader.set_value(f"{config_path}.description", desc)
else:
self.config_loader.set_value(f"{config_path}.host", params["host"])
self.config_loader.set_value(f"{config_path}.port", params["port"])
self.config_loader.set_value(f"{config_path}.user", params["user"])
self.config_loader.set_value(f"{config_path}.password", params["password"])
self.config_loader.set_value(f"{config_path}.name", params["database"])
self.config_loader.set_value(f"{config_path}.description", desc)
# 更新数据源下拉框
self._update_source_combo_items()
# 构建要显示的消息
settings_info = f"数据库类型: {db_type}\n"
for key, value in params.items():
if key != "password":
@ -185,9 +285,15 @@ class SettingsWidget(SettingsUI):
settings_info += f"说明: {desc}"
# 显示成功消息
QMessageBox.information(self, "设置已保存", f"数据库设置已保存!\n\n{settings_info}")
logging.info(f"数据库设置已保存，类型: {db_type}")
except Exception as e:
# 显示错误消息
QMessageBox.critical(self, "保存失败", f"保存数据库设置失败!\n\n错误: {str(e)}")
logging.error(f"保存数据库设置失败: {str(e)}")
def handle_inspection_configs_changed(self):
"""处理检验配置变更"""
logging.info("检验配置已更新")