pythonTest/三一平台.py

import sys
import hashlib
import requests
import json
import time
from datetime import datetime, timedelta
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, 
                             QHBoxLayout, QLabel, QLineEdit, QPushButton, 
                             QTableWidget, QTableWidgetItem, QHeaderView, 
                             QCheckBox, QMessageBox, QGroupBox, QTextEdit,
                             QGridLayout, QSplitter)
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
from PyQt5.QtGui import QTextDocument

class LoginThread(QThread):
    """登录线程"""
    finished = pyqtSignal(dict)
    error = pyqtSignal(str)
    
    def __init__(self, username, password):
        super().__init__()
        self.username = username
        self.password = password
    
    def run(self):
        try:
            md5_password = hashlib.md5(self.password.encode('utf-8')).hexdigest()
            url = f"http://119.91.58.175:9999/gps-web/api/login.jsp?userId={self.username}&password={md5_password}&loginType=user&onlyValidate=false&plateColor=&loginWay=interface&loginLang=zh_CN"
            
            response = requests.get(url, timeout=10)
            response.encoding = 'utf-8'
            
            if response.status_code == 200:
                data = response.json()
                if data.get('rspCode') == '1':
                    self.finished.emit(data)
                else:
                    self.error.emit(f"登录失败: {data.get('msg', '未知错误')}")
            else:
                self.error.emit(f"请求失败: HTTP {response.status_code}")
        except Exception as e:
            self.error.emit(f"登录异常: {str(e)}")

class QueryThread(QThread):
    """查询车辆定位线程"""
    finished = pyqtSignal(dict)
    error = pyqtSignal(str)
    
    def __init__(self, session_id):
        super().__init__()
        self.session_id = session_id
    
    def run(self):
        try:
            url = f"http://119.91.58.175:9999/gps-web/api/get_gps_r.jsp?sessionId={self.session_id}&teamId=&carIds=&simple=false"
            
            response = requests.get(url, timeout=15)
            response.encoding = 'utf-8'
            
            if response.status_code == 200:
                data = response.json()
                if data.get('rspCode') == '1':
                    self.finished.emit(data)
                else:
                    self.error.emit(f"查询失败: {data.get('msg', '未知错误')}")
            else:
                self.error.emit(f"请求失败: HTTP {response.status_code}")
        except Exception as e:
            self.error.emit(f"查询异常: {str(e)}")

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.session_id = None
        self.username = None
        self.car_data = []
        self.car_checked_state = {}  # 存储每辆车的勾选状态
        self.triggered_records = {}  # 存储触发记录 {car_plate: {'load': last_time, 'unload': last_time}}
        self.is_monitoring = False
        self.refresh_timer = QTimer()
        self.refresh_timer.timeout.connect(self.auto_refresh)
        self.initUI()
    
    def initUI(self):
        self.setWindowTitle('车辆定位监控系统')
        self.setGeometry(100, 100, 1400, 800)
        
        # 中央窗口
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        main_layout = QVBoxLayout(central_widget)
        
        # 登录区域
        login_group = QGroupBox("登录")
        login_layout = QHBoxLayout()
        
        login_layout.addWidget(QLabel("账号:"))
        self.username_input = QLineEdit()
        self.username_input.setPlaceholderText("请输入账号")
        self.username_input.setText("深金川")
        login_layout.addWidget(self.username_input)
        
        login_layout.addWidget(QLabel("密码:"))
        self.password_input = QLineEdit()
        self.password_input.setPlaceholderText("请输入密码")
        self.password_input.setEchoMode(QLineEdit.Password)
        self.password_input.setText("123456")
        login_layout.addWidget(self.password_input)
        
        self.login_btn = QPushButton("登录")
        self.login_btn.clicked.connect(self.handle_login)
        login_layout.addWidget(self.login_btn)
        
        self.status_label = QLabel("未登录")
        login_layout.addWidget(self.status_label)
        
        login_layout.addStretch()
        login_group.setLayout(login_layout)
        main_layout.addWidget(login_group)
        
        # 地址输入区域
        addr_group = QGroupBox("地址监控设置")
        addr_layout = QGridLayout()
        
        addr_layout.addWidget(QLabel("装车地址:"), 0, 0)
        self.load_addr_input = QLineEdit()
        self.load_addr_input.setPlaceholderText("请输入装车地址关键词")
        addr_layout.addWidget(self.load_addr_input, 0, 1)
        
        addr_layout.addWidget(QLabel("卸车地址:"), 0, 2)
        self.unload_addr_input = QLineEdit()
        self.unload_addr_input.setPlaceholderText("请输入卸车地址关键词")
        addr_layout.addWidget(self.unload_addr_input, 0, 3)
        
        # 控制按钮
        self.start_btn = QPushButton("启动监控")
        self.start_btn.clicked.connect(self.start_monitoring)
        self.start_btn.setEnabled(False)
        addr_layout.addWidget(self.start_btn, 0, 4)
        
        self.stop_btn = QPushButton("停止监控")
        self.stop_btn.clicked.connect(self.stop_monitoring)
        self.stop_btn.setEnabled(False)
        addr_layout.addWidget(self.stop_btn, 0, 5)
        
        self.manual_query_btn = QPushButton("手动刷新")
        self.manual_query_btn.clicked.connect(self.handle_query)
        self.manual_query_btn.setEnabled(False)
        addr_layout.addWidget(self.manual_query_btn, 0, 6)
        
        addr_layout.addWidget(QLabel("刷新间隔:5秒"), 0, 7)
        
        addr_group.setLayout(addr_layout)
        main_layout.addWidget(addr_group)
        
        # 分割器：表格和日志
        splitter = QSplitter(Qt.Vertical)
        
        # 车辆信息表格
        table_widget = QWidget()
        table_layout = QVBoxLayout(table_widget)
        table_layout.setContentsMargins(0, 0, 0, 0)
        
        self.table = QTableWidget()
        self.table.setColumnCount(5)
        self.table.setHorizontalHeaderLabels(["选择", "车队名", "车牌号", "在线状态", "定位地址"])
        self.table.horizontalHeader().setSectionResizeMode(4, QHeaderView.Stretch)
        self.table.setAlternatingRowColors(True)
        table_layout.addWidget(self.table)
        
        # 表格按钮
        table_btn_layout = QHBoxLayout()
        self.select_all_btn = QPushButton("全选")
        self.select_all_btn.clicked.connect(self.select_all)
        table_btn_layout.addWidget(self.select_all_btn)
        
        self.deselect_all_btn = QPushButton("取消全选")
        self.deselect_all_btn.clicked.connect(self.deselect_all)
        table_btn_layout.addWidget(self.deselect_all_btn)
        
        table_btn_layout.addStretch()
        table_layout.addLayout(table_btn_layout)
        
        splitter.addWidget(table_widget)
        
        # 日志输出框
        log_widget = QWidget()
        log_layout = QVBoxLayout(log_widget)
        log_layout.setContentsMargins(0, 0, 0, 0)
        
        log_label = QLabel("日志输出")
        log_layout.addWidget(log_label)
        
        self.log_output = QTextEdit()
        self.log_output.setReadOnly(True)
        # 限制日志行数 - 使用document().setMaximumBlockCount()
        self.log_output.document().setMaximumBlockCount(1000)
        log_layout.addWidget(self.log_output)
        
        splitter.addWidget(log_widget)
        splitter.setSizes([500, 300])
        
        main_layout.addWidget(splitter)
        
        # 状态栏
        self.statusBar().showMessage("就绪")
        
        # 初始化触发记录
        self.triggered_records = {}
    
    def handle_login(self):
        username = self.username_input.text().strip()
        password = self.password_input.text().strip()
        
        if not username or not password:
            QMessageBox.warning(self, "警告", "请输入账号和密码")
            return
        
        self.login_btn.setEnabled(False)
        self.status_label.setText("登录中...")
        self.statusBar().showMessage("正在登录...")
        
        self.login_thread = LoginThread(username, password)
        self.login_thread.finished.connect(self.on_login_success)
        self.login_thread.error.connect(self.on_login_error)
        self.login_thread.start()
    
    def on_login_success(self, data):
        self.session_id = data.get('sessionId')
        self.username = data.get('userName', '')
        
        if self.session_id:
            self.login_btn.setEnabled(True)
            self.status_label.setText(f"已登录: {self.username}")
            self.manual_query_btn.setEnabled(True)
            self.start_btn.setEnabled(True)
            self.statusBar().showMessage(f"登录成功，Session ID: {self.session_id[:20]}...")
            
            # 登录成功后自动查询
            self.handle_query()
        else:
            self.on_login_error("登录响应中没有sessionId")
    
    def on_login_error(self, error_msg):
        self.login_btn.setEnabled(True)
        self.status_label.setText("登录失败")
        self.statusBar().showMessage("登录失败")
        QMessageBox.critical(self, "错误", error_msg)
    
    def handle_query(self):
        if not self.session_id:
            QMessageBox.warning(self, "警告", "请先登录")
            return
        
        self.manual_query_btn.setEnabled(False)
        self.statusBar().showMessage("正在查询车辆定位...")
        
        self.query_thread = QueryThread(self.session_id)
        self.query_thread.finished.connect(self.on_query_success)
        self.query_thread.error.connect(self.on_query_error)
        self.query_thread.start()
    
    def on_query_success(self, data):
        self.manual_query_btn.setEnabled(True)
        
        car_list = data.get('list', [])
        self.car_data = car_list
        
        # 保存当前勾选状态
        self.save_check_state()
        
        self.table.setRowCount(len(car_list))
        
        for row, car in enumerate(car_list):
            car_plate = car.get('carPlate', '')
            
            # 选择框 - 恢复保存的状态
            checkbox = QCheckBox()
            checkbox.setChecked(self.car_checked_state.get(car_plate, False))
            checkbox.stateChanged.connect(lambda state, plate=car_plate: self.on_check_changed(plate, state))
            self.table.setCellWidget(row, 0, checkbox)
            
            # 车队名
            team_name = car.get('teamName', '')
            self.table.setItem(row, 1, QTableWidgetItem(team_name))
            
            # 车牌号
            self.table.setItem(row, 2, QTableWidgetItem(car_plate))
            
            # 在线状态
            state_cn = car.get('stateCn', '')
            self.table.setItem(row, 3, QTableWidgetItem(state_cn))
            
            # 定位地址
            addr = car.get('addr', '')
            self.table.setItem(row, 4, QTableWidgetItem(addr))
        
        self.statusBar().showMessage(f"查询成功，共 {len(car_list)} 辆车")
        
        # 如果正在监控，进行地址匹配检查
        if self.is_monitoring:
            self.check_address_match()
    
    def on_query_error(self, error_msg):
        self.manual_query_btn.setEnabled(True)
        self.statusBar().showMessage("查询失败")
        QMessageBox.critical(self, "错误", error_msg)
    
    def save_check_state(self):
        """保存当前表格中的勾选状态"""
        for row in range(self.table.rowCount()):
            checkbox = self.table.cellWidget(row, 0)
            if checkbox:
                car_plate = self.table.item(row, 2).text() if self.table.item(row, 2) else ''
                if car_plate:
                    self.car_checked_state[car_plate] = checkbox.isChecked()
    
    def on_check_changed(self, car_plate, state):
        """勾选框状态变化时保存"""
        self.car_checked_state[car_plate] = (state == Qt.Checked)
    
    def select_all(self):
        for row in range(self.table.rowCount()):
            checkbox = self.table.cellWidget(row, 0)
            if checkbox:
                checkbox.setChecked(True)
                car_plate = self.table.item(row, 2).text() if self.table.item(row, 2) else ''
                if car_plate:
                    self.car_checked_state[car_plate] = True
    
    def deselect_all(self):
        for row in range(self.table.rowCount()):
            checkbox = self.table.cellWidget(row, 0)
            if checkbox:
                checkbox.setChecked(False)
                car_plate = self.table.item(row, 2).text() if self.table.item(row, 2) else ''
                if car_plate:
                    self.car_checked_state[car_plate] = False
    
    def start_monitoring(self):
        """启动监控"""
        load_addr = self.load_addr_input.text().strip()
        unload_addr = self.unload_addr_input.text().strip()
        
        if not load_addr and not unload_addr:
            QMessageBox.warning(self, "警告", "请至少输入一个地址关键词")
            return
        
        if not self.session_id:
            QMessageBox.warning(self, "警告", "请先登录")
            return
        
        self.is_monitoring = True
        self.start_btn.setEnabled(False)
        self.stop_btn.setEnabled(True)
        self.manual_query_btn.setEnabled(False)
        
        # 清空触发记录
        self.triggered_records = {}
        
        self.add_log("=" * 50)
        self.add_log(f"监控已启动 - 装车地址: {load_addr or '无'}, 卸车地址: {unload_addr or '无'}")
        self.add_log("=" * 50)
        
        # 立即执行一次检查
        self.handle_query()
        
        # 启动定时器，每5秒刷新一次
        self.refresh_timer.start(5000)
        self.statusBar().showMessage("监控运行中...")
    
    def stop_monitoring(self):
        """停止监控"""
        self.is_monitoring = False
        self.refresh_timer.stop()
        self.start_btn.setEnabled(True)
        self.stop_btn.setEnabled(False)
        self.manual_query_btn.setEnabled(True)
        
        self.add_log("=" * 50)
        self.add_log("监控已停止")
        self.add_log("=" * 50)
        self.statusBar().showMessage("监控已停止")
    
    def auto_refresh(self):
        """自动刷新"""
        if self.is_monitoring and self.session_id:
            self.handle_query()
    
    def check_address_match(self):
        """检查地址匹配"""
        load_keyword = self.load_addr_input.text().strip()
        unload_keyword = self.unload_addr_input.text().strip()
        current_time = datetime.now()
        
        for row in range(self.table.rowCount()):
            car_plate = self.table.item(row, 2).text() if self.table.item(row, 2) else ''
            addr = self.table.item(row, 4).text() if self.table.item(row, 4) else ''
            
            if not car_plate or not addr:
                continue
            
            # 初始化该车辆的触发记录
            if car_plate not in self.triggered_records:
                self.triggered_records[car_plate] = {'load': None, 'unload': None}
            
            # 检查装车地址
            if load_keyword and load_keyword in addr:
                last_trigger = self.triggered_records[car_plate]['load']
                if last_trigger is None or (current_time - last_trigger) >= timedelta(minutes=5):
                    self.triggered_records[car_plate]['load'] = current_time
                    self.add_log(f"🚛 {car_plate} 到达装车位置 ({load_keyword})")
                    self.statusBar().showMessage(f"{car_plate} 到达装车位置")
            
            # 检查卸车地址
            if unload_keyword and unload_keyword in addr:
                last_trigger = self.triggered_records[car_plate]['unload']
                if last_trigger is None or (current_time - last_trigger) >= timedelta(minutes=5):
                    self.triggered_records[car_plate]['unload'] = current_time
                    self.add_log(f"🚛 {car_plate} 到达卸车位置 ({unload_keyword})")
                    self.statusBar().showMessage(f"{car_plate} 到达卸车位置")
    
    def add_log(self, message):
        """添加日志"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_line = f"[{timestamp}] {message}"
        self.log_output.append(log_line)
        
        # 保持最新日志可见
        scrollbar = self.log_output.verticalScrollBar()
        scrollbar.setValue(scrollbar.maximum())

def main():
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec_())

if __name__ == '__main__':
    main()