data-return/waibu.py

600 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import os
import re
import sys
import sqlite3
import subprocess
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QFileDialog, QTableWidget,
QTableWidgetItem, QComboBox, QMessageBox, QHeaderView, QMenu,
QDateEdit, QGroupBox, QRadioButton, QButtonGroup, QSplitter)
from PyQt5.QtCore import Qt, QDate, QThread, pyqtSignal
from PyQt5.QtGui import QIcon, QPixmap
class ExternalDirectoryScanner(QThread):
"""无人机外部检查目录扫描线程"""
scan_finished = pyqtSignal(str, bool, str) # 参数: root_dir, success, message
def __init__(self, root_dir, db_conn, inspector="无人机检查"):
super().__init__()
self.root_dir = root_dir
self.db_conn = db_conn
self.inspector = inspector
def run(self):
try:
cursor = self.db_conn.cursor()
# 清理当前项目路径下的旧数据
print(f"[DEBUG] 清理数据库中的项目路径: {self.root_dir}")
cursor.execute("DELETE FROM external_photos WHERE project_path = ?", (self.root_dir,))
deleted_rows = cursor.rowcount
print(f"[DEBUG] 已删除 {deleted_rows} 条旧记录")
self.db_conn.commit()
if not os.path.exists(self.root_dir):
self.scan_finished.emit(self.root_dir, False, f"目录不存在: {self.root_dir}")
return
# 收集所有机组目录 - 直接使用目录名称作为机组号
unit_dirs = [d for d in os.listdir(self.root_dir)
if os.path.isdir(os.path.join(self.root_dir, d)) and
d != "缺陷"] # 排除缺陷目录
print(f"[DEBUG] 找到 {len(unit_dirs)} 个机组目录: {unit_dirs}")
if not unit_dirs:
self.scan_finished.emit(self.root_dir, False, "未找到任何机组目录")
return
total_photos = 0
for unit_dir in unit_dirs:
unit_path = os.path.join(self.root_dir, unit_dir)
unit_id = unit_dir # 直接使用目录名称作为机组号
print(f"[DEBUG] 正在扫描机组: {unit_id}")
# 首先处理机组目录下的直接图片文件(机组图片)
for file in os.listdir(unit_path):
file_path = os.path.join(unit_path, file)
if os.path.isfile(file_path) and file.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp')):
capture_time = self.parse_capture_time(file)
if capture_time:
cursor.execute("""
INSERT OR IGNORE INTO external_photos (
project_path, unit_id, blade_number, inspector,
photo_path, capture_time, file_name
) VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
self.root_dir, unit_id, "机组", # blade_number设为"机组"表示机组图片
self.inspector,
file_path, capture_time, file
))
if cursor.rowcount > 0:
total_photos += 1
else:
print(f"[DEBUG] 跳过无法解析时间的文件: {file}")
# 然后处理叶片目录
for item in os.listdir(unit_path):
item_path = os.path.join(unit_path, item)
if not os.path.isdir(item_path):
continue
# 直接使用目录名称作为叶片编号
blade_num = item
print(f"[DEBUG] 处理机组 {unit_id}{blade_num}")
for file in os.listdir(item_path):
if file.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp')):
file_path = os.path.join(item_path, file)
capture_time = self.parse_capture_time(file)
if capture_time:
cursor.execute("""
INSERT OR IGNORE INTO external_photos (
project_path, unit_id, blade_number, inspector,
photo_path, capture_time, file_name
) VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
self.root_dir, unit_id, blade_num,
self.inspector,
file_path, capture_time, file
))
if cursor.rowcount > 0:
total_photos += 1
else:
print(f"[DEBUG] 跳过无法解析时间的文件: {file}")
self.db_conn.commit()
print(f"[DEBUG] 扫描完成,共找到 {total_photos} 张照片")
self.scan_finished.emit(self.root_dir, True, f"目录扫描完成,共找到 {total_photos} 张照片")
except Exception as e:
error_msg = f"扫描目录时出错: {str(e)}"
print(f"[ERROR] {error_msg}")
self.scan_finished.emit(self.root_dir, False, error_msg)
def parse_capture_time(self, filename):
"""从文件名解析拍摄时间"""
# 尝试第一种格式DJI_YYYYMMDDHHMMSS_xxx_Z.JPG
match = re.search(r"DJI_(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})", filename)
if match:
year, month, day, hour, minute, second = match.groups()
try:
return f"{year}-{month}-{day} {hour}:{minute}:{second}"
except:
return None
# 尝试第二种格式YYYYMMDD_HHMMSS.jpg
match = re.search(r"(\d{4})(\d{2})(\d{2})_(\d{2})(\d{2})(\d{2})", filename)
if match:
year, month, day, hour, minute, second = match.groups()
try:
return f"{year}-{month}-{day} {hour}:{minute}:{second}"
except:
return None
return None
class ExternalPhotoManager(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("风电叶片无人机外部检查照片管理系统")
self.setWindowIcon(QIcon("drone.ico"))
self.setGeometry(100, 100, 1200, 800)
# 使用独立的外部检查数据库
self.db_conn = sqlite3.connect("external_inspection.db",
check_same_thread=False)
self.create_db_table()
# 初始化UI组件
self.init_ui_components()
# 加载数据
self.load_units_to_combo()
self.show()
def create_db_table(self):
"""创建独立的外部检查数据库表结构"""
cursor = self.db_conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS external_photos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_path TEXT NOT NULL,
unit_id TEXT NOT NULL,
blade_number TEXT NOT NULL,
inspector TEXT NOT NULL DEFAULT '无人机检查',
photo_path TEXT NOT NULL UNIQUE,
capture_time TEXT NOT NULL,
file_name TEXT NOT NULL
)
""")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_ext_project_path ON external_photos(project_path)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_ext_unit_id ON external_photos(unit_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_ext_blade_number ON external_photos(blade_number)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_ext_capture_time ON external_photos(capture_time)")
self.db_conn.commit()
def init_ui_components(self):
"""初始化无人机外部检查专用UI"""
main_widget = QWidget()
main_layout = QVBoxLayout()
# 目录选择区域
dir_layout = QHBoxLayout()
self.dir_label = QLabel("项目目录:")
self.dir_input = QLineEdit()
self.dir_input.setPlaceholderText("请选择无人机外部检查照片根目录")
self.browse_btn = QPushButton("浏览...")
self.browse_btn.clicked.connect(self.browse_directory)
self.scan_btn = QPushButton("扫描目录")
self.scan_btn.clicked.connect(self.start_scan_directory)
dir_layout.addWidget(self.dir_label)
dir_layout.addWidget(self.dir_input, stretch=1)
dir_layout.addWidget(self.browse_btn)
dir_layout.addWidget(self.scan_btn)
# 查询条件区域
query_layout = QHBoxLayout()
# 机组编号查询
unit_layout = QVBoxLayout()
unit_layout.addWidget(QLabel("机组编号"))
self.unit_combo = QComboBox()
self.unit_combo.addItem("所有机组", "")
unit_layout.addWidget(self.unit_combo)
# 检查类型选择
type_layout = QVBoxLayout()
type_layout.addWidget(QLabel("检查类型"))
self.type_group = QButtonGroup()
self.all_types_radio = QRadioButton("所有类型")
self.blade_radio = QRadioButton("叶片")
self.tower_radio = QRadioButton("塔筒")
self.unit_radio = QRadioButton("机组图片")
self.all_types_radio.setChecked(True)
self.type_group.addButton(self.all_types_radio)
self.type_group.addButton(self.blade_radio)
self.type_group.addButton(self.tower_radio)
self.type_group.addButton(self.unit_radio)
type_layout.addWidget(self.all_types_radio)
type_layout.addWidget(self.blade_radio)
type_layout.addWidget(self.tower_radio)
type_layout.addWidget(self.unit_radio)
# 日期范围查询
date_group = QGroupBox("日期范围")
date_layout = QHBoxLayout()
self.start_date_edit = QDateEdit()
self.start_date_edit.setCalendarPopup(True)
self.start_date_edit.setDisplayFormat("yyyy-MM-dd")
self.start_date_edit.setDate(QDate.currentDate().addDays(-7))
self.end_date_edit = QDateEdit()
self.end_date_edit.setCalendarPopup(True)
self.end_date_edit.setDisplayFormat("yyyy-MM-dd")
self.end_date_edit.setDate(QDate.currentDate())
date_layout.addWidget(self.start_date_edit)
date_layout.addWidget(QLabel(""))
date_layout.addWidget(self.end_date_edit)
date_group.setLayout(date_layout)
# 叶片编号选择框
self.blade_combo = QComboBox()
self.blade_combo.addItem("所有叶片", "")
self.blade_combo.setEnabled(False)
# 将叶片编号添加到机组布局中
unit_layout.addWidget(QLabel("叶片编号"))
unit_layout.addWidget(self.blade_combo)
query_layout.addLayout(unit_layout)
query_layout.addLayout(type_layout)
query_layout.addWidget(date_group)
# 查询按钮
self.query_btn = QPushButton("查询")
self.query_btn.clicked.connect(self.query_photos)
query_layout.addWidget(self.query_btn)
# 主内容区域 - 使用分割器确保预览区域可见
splitter = QSplitter(Qt.Horizontal)
# 结果显示区域
self.result_table = QTableWidget()
self.result_table.setColumnCount(6) # 初始设置为最大列数
self.result_table.setHorizontalHeaderLabels(["机组编号", "叶片/塔筒编号", "检查类型", "拍摄时间", "文件名", "路径"])
self.result_table.setSelectionBehavior(QTableWidget.SelectRows)
self.result_table.setEditTriggers(QTableWidget.NoEditTriggers)
self.result_table.doubleClicked.connect(self.open_selected_photo)
# 设置列宽策略
header = self.result_table.horizontalHeader()
header.setSectionResizeMode(0, QHeaderView.ResizeToContents)
header.setSectionResizeMode(1, QHeaderView.ResizeToContents)
header.setSectionResizeMode(2, QHeaderView.ResizeToContents)
header.setSectionResizeMode(3, QHeaderView.ResizeToContents)
header.setSectionResizeMode(4, QHeaderView.Stretch)
header.setSectionResizeMode(5, QHeaderView.Stretch)
# 预览区域
preview_widget = QWidget()
preview_layout = QVBoxLayout()
preview_layout.addWidget(QLabel("照片预览"))
self.preview_label = QLabel()
self.preview_label.setAlignment(Qt.AlignCenter)
self.preview_label.setMinimumSize(300, 300)
self.preview_label.setText("照片预览区域")
self.preview_label.setStyleSheet("""
QLabel {
border: 1px solid gray;
background-color: #f0f0f0;
qproperty-alignment: AlignCenter;
}
""")
preview_layout.addWidget(self.preview_label)
preview_widget.setLayout(preview_layout)
# 将表格和预览区域添加到分割器
splitter.addWidget(self.result_table)
splitter.addWidget(preview_widget)
splitter.setStretchFactor(0, 3) # 表格占3份空间
splitter.setStretchFactor(1, 1) # 预览占1份空间
# 连接信号
self.type_group.buttonClicked.connect(self.update_blade_combo)
self.unit_combo.currentIndexChanged.connect(self.update_blade_combo)
# 连接选择变化信号
self.result_table.selectionModel().selectionChanged.connect(self.update_preview)
# 状态栏
self.statusBar().showMessage("就绪")
# 布局组装
main_layout.addLayout(dir_layout)
main_layout.addLayout(query_layout)
main_layout.addWidget(splitter, stretch=1) # 使用分割器作为主要内容区域
main_widget.setLayout(main_layout)
self.setCentralWidget(main_widget)
def browse_directory(self):
dir_path = QFileDialog.getExistingDirectory(self, "选择无人机外部检查照片根目录")
if dir_path:
self.dir_input.setText(dir_path)
# 自动加载该目录下的机组
self.load_units_to_combo(dir_path)
def update_blade_combo(self):
"""更新叶片编号下拉框"""
if self.blade_radio.isChecked():
unit_id = self.unit_combo.currentData()
self.blade_combo.setEnabled(True)
cursor = self.db_conn.cursor()
query = """
SELECT DISTINCT blade_number
FROM external_photos
WHERE unit_id = ? AND blade_number NOT IN ('塔筒', '机组')
AND project_path = ?
ORDER BY blade_number
"""
cursor.execute(query, (unit_id, self.dir_input.text()))
blades = [row[0] for row in cursor.fetchall()]
self.blade_combo.clear()
self.blade_combo.addItem("所有叶片", "")
for blade in blades:
clean_blade = blade.replace("#", "")
self.blade_combo.addItem(clean_blade, blade)
else:
self.blade_combo.setEnabled(False)
def start_scan_directory(self):
"""启动目录扫描线程"""
root_dir = self.dir_input.text()
if not root_dir or not os.path.exists(root_dir):
QMessageBox.warning(self, "警告", "请选择有效的目录路径")
return
self.scan_btn.setEnabled(False)
self.statusBar().showMessage("正在扫描目录,请稍候...")
self.scanner = ExternalDirectoryScanner(root_dir, self.db_conn)
self.scanner.scan_finished.connect(self.on_scan_finished)
self.scanner.start()
def on_scan_finished(self, root_dir, success, message):
"""扫描完成后的回调"""
self.scan_btn.setEnabled(True)
if success:
self.load_units_to_combo(root_dir)
self.query_photos()
self.statusBar().showMessage(f"目录扫描完成: {root_dir}", 5000)
QMessageBox.information(self, "完成", message)
else:
error_msg = f"扫描目录时出错:\n\n{message}\n\n请检查:\n1. 目录结构是否符合要求\n2. 是否有访问权限"
QMessageBox.critical(self, "扫描错误", error_msg)
self.statusBar().showMessage(f"扫描失败: {message}", 5000)
def load_units_to_combo(self, project_path=None):
"""加载机组编号到下拉框"""
cursor = self.db_conn.cursor()
# 如果没有传入project_path尝试使用当前目录输入框的值
if project_path is None:
project_path = self.dir_input.text() if self.dir_input.text() else None
if project_path:
cursor.execute("""
SELECT DISTINCT unit_id FROM external_photos
WHERE unit_id != '' AND project_path = ?
ORDER BY unit_id
""", (project_path,))
else:
cursor.execute("""
SELECT DISTINCT unit_id FROM external_photos
WHERE unit_id != ''
ORDER BY unit_id
""")
units = [row[0] for row in cursor.fetchall()]
self.unit_combo.clear()
self.unit_combo.addItem("所有机组", "")
for unit in units:
self.unit_combo.addItem(unit, unit)
def query_photos(self):
"""查询照片数据"""
# 获取查询条件
unit_id = self.unit_combo.currentData() or None
start_date = self.start_date_edit.date().toString("yyyy-MM-dd")
end_date = self.end_date_edit.date().toString("yyyy-MM-dd")
project_path = self.dir_input.text() # 获取当前扫描目录
# 根据检查类型设置表格列
is_blade = self.blade_radio.isChecked()
# 设置表格列标题
if is_blade:
# 叶片模式: 显示叶片编号列
self.result_table.setColumnCount(6)
headers = ["机组编号", "叶片编号", "检查类型", "拍摄时间", "文件名", "路径"]
else:
# 非叶片模式: 隐藏叶片编号列
self.result_table.setColumnCount(5)
headers = ["机组编号", "检查类型", "拍摄时间", "文件名", "路径"]
self.result_table.setHorizontalHeaderLabels(headers)
# 构建基础查询
query = """
SELECT
unit_id,
blade_number,
CASE
WHEN blade_number = '塔筒' THEN '塔筒'
WHEN blade_number = '机组' THEN '机组图片'
ELSE '叶片'
END AS inspection_type,
capture_time,
file_name,
photo_path
FROM external_photos
WHERE date(capture_time) BETWEEN ? AND ?
AND project_path = ?
"""
params = [start_date, end_date, project_path]
# 添加机组条件
if unit_id:
query += " AND unit_id = ?"
params.append(unit_id)
# 添加检查类型条件
if is_blade:
query += " AND blade_number NOT IN ('塔筒', '机组')"
# 添加叶片编号条件
blade_num = self.blade_combo.currentData()
if blade_num:
query += " AND blade_number = ?"
params.append(blade_num)
elif self.tower_radio.isChecked():
query += " AND blade_number = '塔筒'"
elif self.unit_radio.isChecked():
query += " AND blade_number = '机组'"
query += " ORDER BY unit_id, blade_number, capture_time"
# 执行查询
cursor = self.db_conn.cursor()
try:
cursor.execute(query, params)
results = cursor.fetchall()
# 处理结果显示
processed_results = []
for row in results:
unit, blade, typ, time, name, path = row
if is_blade:
# 叶片模式下显示叶片编号(去除#号)
clean_blade = blade.replace("#", "")
processed_results.append((unit, clean_blade, typ, time, name, path))
else:
# 非叶片模式下不显示叶片编号
processed_results.append((unit, typ, time, name, path))
self.result_table.setRowCount(len(processed_results))
for row_idx, row in enumerate(processed_results):
for col_idx, col in enumerate(row):
item = QTableWidgetItem(str(col) if col is not None else "")
self.result_table.setItem(row_idx, col_idx, item)
self.statusBar().showMessage(f"找到 {len(processed_results)} 条记录", 5000)
# 如果有结果,自动选择第一行并更新预览
if processed_results:
self.result_table.selectRow(0)
self.update_preview()
else:
self.preview_label.setText("无照片可预览")
self.preview_label.setPixmap(QPixmap())
except sqlite3.Error as e:
QMessageBox.critical(self, "数据库错误", f"查询失败: {str(e)}")
def open_selected_photo(self, index):
"""双击打开选中的照片"""
selected_row = index.row()
# 根据当前显示模式确定路径列索引
is_blade = self.blade_radio.isChecked()
path_col = 5 if is_blade else 4
photo_path = self.result_table.item(selected_row, path_col).text()
if not photo_path:
QMessageBox.warning(self, "警告", "未找到文件路径")
return
photo_path = os.path.normpath(photo_path)
if not os.path.exists(photo_path):
QMessageBox.warning(self, "文件不存在", f"无法找到文件:\n{photo_path}")
return
try:
if os.name == 'nt':
os.startfile(photo_path)
elif sys.platform == 'darwin':
subprocess.call(['open', photo_path])
else:
subprocess.call(['xdg-open', photo_path])
self.statusBar().showMessage(f"正在打开: {os.path.basename(photo_path)}", 3000)
except Exception as e:
QMessageBox.critical(self, "错误", f"无法打开文件:\n路径: {photo_path}\n错误: {str(e)}")
def update_preview(self):
"""更新照片预览"""
selected_rows = self.result_table.selectionModel().selectedRows()
if selected_rows:
selected_row = selected_rows[0].row()
# 根据当前显示模式确定路径列索引
is_blade = self.blade_radio.isChecked()
path_col = 5 if is_blade else 4
photo_path = self.result_table.item(selected_row, path_col).text()
if photo_path and os.path.exists(photo_path):
try:
pixmap = QPixmap(photo_path)
if not pixmap.isNull():
# 保持宽高比缩放图片
scaled_pixmap = pixmap.scaled(
self.preview_label.width(),
self.preview_label.height(),
Qt.KeepAspectRatio,
Qt.SmoothTransformation
)
self.preview_label.setPixmap(scaled_pixmap)
self.preview_label.setText("")
else:
self.preview_label.setText("无法加载预览")
self.preview_label.setPixmap(QPixmap())
except Exception as e:
print(f"预览错误: {str(e)}")
self.preview_label.setText("预览错误")
self.preview_label.setPixmap(QPixmap())
else:
self.preview_label.setText("文件不存在")
self.preview_label.setPixmap(QPixmap())
def closeEvent(self, event):
self.db_conn.close()
event.accept()
if __name__ == "__main__":
app = QApplication(sys.argv)
manager = ExternalPhotoManager()
sys.exit(app.exec_())