413 lines
15 KiB
Python
413 lines
15 KiB
Python
import logging
|
||
from datetime import datetime
|
||
from PySide6.QtWidgets import QTableWidgetItem, QMessageBox
|
||
from PySide6.QtCore import Qt, Signal
|
||
from ui.luno_query_dialog_ui import LunoQueryDialogUI
|
||
from utils.sql_utils import SQLUtils
|
||
from utils.app_mode import AppMode
|
||
from apis.gc_api import GcApi
|
||
|
||
class LunoQueryDialog(LunoQueryDialogUI):
|
||
"""炉号查询对话框，用于查询炉号信息"""
|
||
|
||
# 定义信号，用于向上传递选中的炉号信息
|
||
luno_selected = Signal(dict)
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__()
|
||
|
||
# 存储查询结果
|
||
self.query_results = []
|
||
|
||
# 获取父窗口的用户信息
|
||
self.corp_id = None
|
||
self.user_id = None
|
||
self.user_name = None
|
||
if parent:
|
||
if hasattr(parent, 'corp_id'):
|
||
self.corp_id = parent.corp_id
|
||
if hasattr(parent, 'user_id'):
|
||
self.user_id = parent.user_id[0] if isinstance(parent.user_id, tuple) else parent.user_id
|
||
if hasattr(parent, 'user_name'):
|
||
self.user_name = parent.user_name
|
||
|
||
# 连接信号和槽
|
||
self.connect_signals()
|
||
|
||
def connect_signals(self):
|
||
"""连接信号和槽"""
|
||
# 查询按钮点击事件
|
||
self.query_button.clicked.connect(self.on_luno_query)
|
||
|
||
# 确认按钮点击事件
|
||
self.confirm_button.clicked.connect(self.on_confirm)
|
||
|
||
# 取消按钮点击事件
|
||
self.cancel_button.clicked.connect(self.reject)
|
||
|
||
# 表格双击事件
|
||
self.result_table.cellDoubleClicked.connect(self.on_table_double_clicked)
|
||
|
||
def on_luno_query(self):
|
||
"""执行炉号查询"""
|
||
try:
|
||
# 获取查询参数
|
||
start_date = self.start_date.date().toString("yyyy-MM-dd")
|
||
end_date = self.end_date.date().toString("yyyy-MM-dd")
|
||
luno = self.luno_input.text().strip()
|
||
material = self.material_combo.currentText()
|
||
if material == "全部":
|
||
material = ""
|
||
spec = self.spec_input.text().strip()
|
||
steel = self.steel_input.text().strip()
|
||
|
||
# 构建查询参数
|
||
query_params = {
|
||
"srch_rq1": start_date,
|
||
"srch_rq2": end_date,
|
||
"luono": luno,
|
||
"cz": material,
|
||
"gg": spec,
|
||
"gc": steel,
|
||
"corp_id": self.corp_id
|
||
}
|
||
|
||
# 执行查询
|
||
results = self.query_lunos(query_params)
|
||
|
||
# 更新表格
|
||
self.update_result_table(results)
|
||
|
||
except Exception as e:
|
||
logging.error(f"炉号查询失败: {e}")
|
||
QMessageBox.critical(self, "查询错误", f"查询炉号时发生错误: {str(e)}")
|
||
|
||
def query_lunos(self, params):
|
||
"""查询炉号数据
|
||
|
||
Args:
|
||
params: 查询参数字典
|
||
|
||
Returns:
|
||
list: 炉号数据列表
|
||
"""
|
||
try:
|
||
# 如果是API模式，优先从接口获取数据
|
||
if AppMode.is_api():
|
||
return self._query_lunos_from_api(params)
|
||
else:
|
||
return self._query_lunos_from_db(params)
|
||
|
||
except Exception as e:
|
||
logging.error(f"查询炉号数据失败: {e}")
|
||
raise
|
||
|
||
def _query_lunos_from_api(self, params):
|
||
"""从API获取炉号数据
|
||
|
||
Args:
|
||
params: 查询参数字典
|
||
|
||
Returns:
|
||
list: 炉号数据列表
|
||
"""
|
||
try:
|
||
# 调用接口
|
||
gc_api = GcApi()
|
||
|
||
# 调用接口查询炉号列表
|
||
response = gc_api.get_luno_list(params)
|
||
|
||
# 处理接口返回结果 - 接口直接返回数据列表
|
||
if response and isinstance(response, list):
|
||
# 接口直接返回炉号数据列表
|
||
lunos_data = response
|
||
results = []
|
||
|
||
for luno_data in lunos_data:
|
||
# 转换为统一格式的字典，处理接口返回的实际字段
|
||
luno_dict = {
|
||
"luono": luno_data.get("luono", ""),
|
||
"cz": luno_data.get("cz", ""),
|
||
"gg": luno_data.get("gg", ""),
|
||
"gc": luno_data.get("gc", ""),
|
||
"bz": luno_data.get("bz", ""),
|
||
"data_corp": luno_data.get("data_corp", ""),
|
||
"data_corp_name": luno_data.get("data_corp_name", ""),
|
||
"c": luno_data.get("c", ""),
|
||
"si": luno_data.get("si", ""),
|
||
"mn": luno_data.get("mn", ""),
|
||
"p": luno_data.get("p", ""),
|
||
"s": luno_data.get("s", ""),
|
||
"ni": luno_data.get("ni", ""),
|
||
"cr": luno_data.get("cr", ""),
|
||
"ti": luno_data.get("ti", ""),
|
||
"mo": luno_data.get("mo", ""),
|
||
"cu": luno_data.get("cu", ""),
|
||
"others": luno_data.get("others", ""),
|
||
"klqd": luno_data.get("klqd", ""),
|
||
"ysl": luno_data.get("ysl", ""),
|
||
"rq": luno_data.get("rq", "")
|
||
}
|
||
|
||
# 处理null值，转换为空字符串
|
||
for key, value in luno_dict.items():
|
||
if value is None:
|
||
luno_dict[key] = ""
|
||
|
||
results.append(luno_dict)
|
||
|
||
# 保存查询结果
|
||
self.query_results = results
|
||
|
||
# 如果接口查询到数据，则保存到数据库中
|
||
if results:
|
||
self._save_lunos_to_db(results)
|
||
|
||
return results
|
||
else:
|
||
# 如果接口返回的不是列表或为空，则尝试从数据库查询
|
||
logging.warning("从API获取炉号数据失败或返回格式不正确，尝试从数据库查询")
|
||
return self._query_lunos_from_db(params)
|
||
|
||
except Exception as e:
|
||
logging.error(f"从API获取炉号数据失败: {e}")
|
||
# 如果接口查询失败，则尝试从数据库查询
|
||
logging.warning("从API获取炉号数据失败，尝试从数据库查询")
|
||
return self._query_lunos_from_db(params)
|
||
|
||
def _save_lunos_to_db(self, lunos):
|
||
"""将炉号数据保存到数据库
|
||
|
||
Args:
|
||
lunos: 炉号数据列表
|
||
"""
|
||
try:
|
||
# 导入DAO
|
||
from dao.inspection_dao import InspectionDAO
|
||
inspection_dao = InspectionDAO()
|
||
|
||
# 遍历炉号数据，保存到数据库
|
||
for luno_data in lunos:
|
||
luno_code = luno_data.get("luono", "")
|
||
if luno_code:
|
||
# 设置用户信息
|
||
if self.user_id:
|
||
luno_data['user_id'] = self.user_id
|
||
if self.user_name:
|
||
luno_data['user_name'] = self.user_name
|
||
if self.corp_id:
|
||
luno_data['data_corp'] = self.corp_id
|
||
|
||
# 保存炉号信息
|
||
inspection_dao.save_luno_info(luno_code, luno_data)
|
||
|
||
logging.info(f"成功保存{len(lunos)}条炉号数据到数据库")
|
||
|
||
except Exception as e:
|
||
logging.error(f"保存炉号数据到数据库失败: {e}")
|
||
|
||
def _query_lunos_from_db(self, params):
|
||
"""从数据库获取炉号数据
|
||
|
||
Args:
|
||
params: 查询参数字典
|
||
|
||
Returns:
|
||
list: 炉号数据列表
|
||
"""
|
||
# 构建SQL查询
|
||
sql = """
|
||
SELECT
|
||
l.luono, l.cz, l.gg, l.gc, l.bz, l.data_corp, l.data_corp_name,
|
||
l.c, l.si, l.mn, l.p, l.s, l.ni, l.cr, l.ti, l.mo, l.cu, l.others,
|
||
l.klqd, l.ysl, l.create_time as rq
|
||
FROM wsbz_luno_info l
|
||
WHERE 1=1
|
||
"""
|
||
|
||
query_params = []
|
||
|
||
# 添加查询条件
|
||
if params.get("srch_rq1") and params.get("srch_rq2"):
|
||
sql += " AND l.create_time BETWEEN ? AND ?"
|
||
query_params.append(params["srch_rq1"] + " 00:00:00")
|
||
query_params.append(params["srch_rq2"] + " 23:59:59")
|
||
|
||
if params.get("luono"):
|
||
sql += " AND l.luono LIKE ?"
|
||
query_params.append(f"%{params['luono']}%")
|
||
|
||
if params.get("cz"):
|
||
sql += " AND l.cz = ?"
|
||
query_params.append(params["cz"])
|
||
|
||
if params.get("gg"):
|
||
sql += " AND l.gg LIKE ?"
|
||
query_params.append(f"%{params['gg']}%")
|
||
|
||
if params.get("gc"):
|
||
sql += " AND l.gc LIKE ?"
|
||
query_params.append(f"%{params['gc']}%")
|
||
|
||
# 添加排序
|
||
sql += " ORDER BY l.create_time DESC"
|
||
|
||
# 执行查询
|
||
with SQLUtils('sqlite') as db:
|
||
db.execute_query(sql, query_params)
|
||
rows = db.fetchall()
|
||
|
||
# 处理查询结果
|
||
results = []
|
||
for row in rows:
|
||
# 将元组转换为字典
|
||
luno_data = {
|
||
"luono": row[0],
|
||
"cz": row[1],
|
||
"gg": row[2],
|
||
"gc": row[3],
|
||
"bz": row[4],
|
||
"data_corp": row[5],
|
||
"data_corp_name": row[6],
|
||
"c": row[7],
|
||
"si": row[8],
|
||
"mn": row[9],
|
||
"p": row[10],
|
||
"s": row[11],
|
||
"ni": row[12],
|
||
"cr": row[13],
|
||
"ti": row[14],
|
||
"mo": row[15],
|
||
"cu": row[16],
|
||
"others": row[17],
|
||
"klqd": row[18],
|
||
"ysl": row[19],
|
||
"rq": row[20]
|
||
}
|
||
results.append(luno_data)
|
||
|
||
# 保存查询结果
|
||
self.query_results = results
|
||
|
||
return results
|
||
|
||
def update_result_table(self, results):
|
||
"""更新结果表格
|
||
|
||
Args:
|
||
results: 炉号数据列表
|
||
"""
|
||
# 清空表格
|
||
self.result_table.setRowCount(0)
|
||
|
||
if not results:
|
||
return
|
||
|
||
# 定义表头和对应的字段名
|
||
columns = [
|
||
{"title": "序号", "field": None},
|
||
{"title": "日期", "field": "rq"},
|
||
{"title": "炉号", "field": "luono"},
|
||
{"title": "材质", "field": "cz"},
|
||
{"title": "规格", "field": "gg"},
|
||
{"title": "钢厂", "field": "gc"},
|
||
{"title": "标准", "field": "bz"},
|
||
{"title": "公司账套", "field": "data_corp_name"},
|
||
{"title": "C", "field": "c"},
|
||
{"title": "Si", "field": "si"},
|
||
{"title": "Mn", "field": "mn"},
|
||
{"title": "P", "field": "p"},
|
||
{"title": "S", "field": "s"},
|
||
{"title": "Ni", "field": "ni"},
|
||
{"title": "Cr", "field": "cr"},
|
||
{"title": "Ti", "field": "ti"},
|
||
{"title": "Mo", "field": "mo"},
|
||
{"title": "Cu", "field": "cu"},
|
||
{"title": "其他", "field": "others"},
|
||
{"title": "抗拉强度", "field": "klqd"},
|
||
{"title": "延伸率", "field": "ysl"}
|
||
]
|
||
|
||
# 设置表头
|
||
self.result_table.setColumnCount(len(columns))
|
||
header_labels = [col["title"] for col in columns]
|
||
self.result_table.setHorizontalHeaderLabels(header_labels)
|
||
|
||
# 设置行数
|
||
self.result_table.setRowCount(len(results))
|
||
|
||
# 填充数据
|
||
for row, luno_data in enumerate(results):
|
||
# 序号
|
||
item_index = QTableWidgetItem(str(row + 1))
|
||
item_index.setTextAlignment(Qt.AlignCenter)
|
||
self.result_table.setItem(row, 0, item_index)
|
||
|
||
# 遍历列，填充数据
|
||
for col_idx, column in enumerate(columns[1:], 1): # 从1开始，跳过序号列
|
||
field = column["field"]
|
||
if field:
|
||
value = ""
|
||
|
||
# 特殊处理某些字段
|
||
if field == "rq":
|
||
create_time = luno_data.get(field, "")
|
||
if create_time:
|
||
try:
|
||
# 尝试解析日期时间字符串，支持多种格式
|
||
if " " in create_time:
|
||
# 包含时间的格式：2025-07-09 10:30:00
|
||
dt = datetime.strptime(create_time, "%Y-%m-%d %H:%M:%S")
|
||
value = dt.strftime("%Y-%m-%d")
|
||
else:
|
||
# 只有日期的格式：2025-07-09
|
||
dt = datetime.strptime(create_time, "%Y-%m-%d")
|
||
value = dt.strftime("%Y-%m-%d")
|
||
except:
|
||
# 如果解析失败，直接使用原值
|
||
value = create_time
|
||
else:
|
||
# 其他普通字段
|
||
value = str(luno_data.get(field, ""))
|
||
|
||
# 创建表格项并设置文本
|
||
item = QTableWidgetItem(value)
|
||
|
||
# 居中对齐特定的列
|
||
if field in ["rq", "c", "si", "mn", "p", "s", "ni", "cr", "ti", "mo", "cu", "klqd", "ysl"]:
|
||
item.setTextAlignment(Qt.AlignCenter)
|
||
|
||
# 设置表格项
|
||
self.result_table.setItem(row, col_idx, item)
|
||
|
||
# 存储原始数据到第一列的item中
|
||
item_index.setData(Qt.UserRole, luno_data)
|
||
|
||
def on_table_double_clicked(self, row, column):
|
||
"""表格双击事件处理"""
|
||
self.select_current_row(row)
|
||
|
||
def on_confirm(self):
|
||
"""确认按钮点击事件处理"""
|
||
# 获取当前选中行
|
||
selected_rows = self.result_table.selectionModel().selectedRows()
|
||
if not selected_rows:
|
||
QMessageBox.warning(self, "提示", "请选择一个炉号")
|
||
return
|
||
|
||
# 获取选中行的索引
|
||
row_index = selected_rows[0].row()
|
||
self.select_current_row(row_index)
|
||
|
||
def select_current_row(self, row):
|
||
"""选择当前行并返回数据"""
|
||
if 0 <= row < self.result_table.rowCount():
|
||
# 获取存储在item中的原始数据
|
||
item = self.result_table.item(row, 0)
|
||
if item:
|
||
luno_data = item.data(Qt.UserRole)
|
||
if luno_data:
|
||
# 发出信号
|
||
self.luno_selected.emit(luno_data)
|
||
self.accept() |