Feeding_control_system/tcp/server.py

import socket
import json
import threading
import time



class TCPServer:
    def __init__(self, host='127.0.0.1', port=8888):
        self.host = host
        self.port = port
        self.server_socket = None
        self.is_running = False
        self.client_sockets = []

    def start(self):
        """启动TCP服务端"""
        try:
            self.is_running = True
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.server_socket.bind((self.host, self.port))
            self.server_socket.listen(5)
            print(f"TCP服务端已启动，监听 {self.host}:{self.port}...")

            # 启动接受连接的线程
            accept_thread = threading.Thread(target=self.accept_connections, daemon=True)
            accept_thread.start()

            # 保持主线程运行
            try:
                while self.is_running:
                    time.sleep(1)
            except KeyboardInterrupt:
                print("\n服务端正在关闭...")
                self.stop()

        except Exception as e:
            print(f"启动服务端时出错: {e}")

    def accept_connections(self):
        """接受客户端连接"""
        while self.is_running:
            try:
                client_socket, client_address = self.server_socket.accept()
                self.client_sockets.append(client_socket)
                print(f"客户端 {client_address} 已连接")

                # 启动一个线程处理客户端通信
                client_thread = threading.Thread(
                    target=self.handle_client,
                    args=(client_socket, client_address),
                    daemon=True
                )
                client_thread.start()

            except Exception as e:
                if self.is_running:
                    print(f"接受客户端连接时出错: {e}")
                break

    def handle_client(self, client_socket, client_address):
        """处理客户端通信"""
        try:
            while self.is_running and client_socket in self.client_sockets:
                try:
                    # 设置接收超时
                    client_socket.settimeout(1.0)

                    # 接收客户端数据
                    data = client_socket.recv(1024)
                    if not data:
                        break

                    # 解析JSON数据
                    message = data.decode('utf-8').strip()
                    if message:
                        json_data = json.loads(message)
                        # 处理客户端返回的异常数据
                        self.process_client_data(json_data)

                except socket.timeout:
                    # 超时继续循环
                    continue
                except json.JSONDecodeError as e:
                    print(f"解析客户端 {client_address} JSON数据时出错: {e}")
                except Exception as e:
                    print(f"处理客户端 {client_address} 数据时出错: {e}")
                    break

        except Exception as e:
            print(f"客户端 {client_address} 处理线程出错: {e}")
        finally:
            # 清理客户端连接
            if client_socket in self.client_sockets:
                self.client_sockets.remove(client_socket)
            client_socket.close()
            print(f"客户端 {client_address} 已断开连接")

    def process_client_data(self, json_data):
        """处理客户端返回的数据"""
        try:
            cmd = json_data.get("cmd")
            timestamp = json_data.get("timestamp")
            erp_id = json_data.get("erp_id")  # 从JSON数据中解析ErpID

            if cmd == "production_error":
                print(f"收到异常生产通知 - 时间: {timestamp}, ERP ID: {erp_id}")
                # 调用update_custom_table_status更新数据库状态
                try:
                    from services.task_service import TaskService
                    task_service = TaskService()
                    task_service.update_custom_table_status(erp_id, "异常生产")
                    print(f"数据库状态已更新为: 异常生产 (ERP ID: {erp_id})")
                except Exception as e:
                    print(f"更新数据库状态时出错: {e}")
            elif cmd == "cancel_feed":
                print(f"收到中断生产通知 - 时间: {timestamp}, ERP ID: {erp_id}")
                # 调用update_custom_table_status更新数据库状态
                try:
                    from services.task_service import TaskService
                    task_service = TaskService()
                    task_service.update_custom_table_status(erp_id, "取消生产")
                    print(f"数据库状态已更新为: 取消生产 (ERP ID: {erp_id})")
                except Exception as e:
                    print(f"更新数据库状态时出错: {e}")
            else:
                print(f"收到未知命令: {cmd}")

        except Exception as e:
            print(f"处理客户端数据时出错: {e}")

    def send_data(self, data):
        """向所有已连接的客户端发送数据"""
        disconnected_sockets = []

        for client_socket in self.client_sockets:
            try:
                if client_socket:
                    json_data = json.dumps(data, ensure_ascii=False)
                    # 添加换行符作为结束标记
                    client_socket.sendall((json_data + "\n").encode('utf-8'))
                    print(f"已向客户端发送数据: {json_data}")
            except Exception as e:
                print(f"向客户端发送数据时出错: {e}")
                disconnected_sockets.append(client_socket)

        # 清理断开连接的套接字
        for sock in disconnected_sockets:
            if sock in self.client_sockets:
                self.client_sockets.remove(sock)

    def stop(self):
        """停止服务端"""
        self.is_running = False

        # 关闭所有客户端连接
        for sock in self.client_sockets:
            try:
                sock.close()
            except Exception as e:
                print(f"关闭客户端连接时发生错误: {e}")

        # 关闭服务器套接字
        if self.server_socket:
            try:
                self.server_socket.close()
            except Exception as e:
                print(f"关闭服务器套接字时发生错误: {e}")

        print("TCP服务端已停止")
        self.client_sockets.clear()


# 使用示例
if __name__ == '__main__':
    server = TCPServer(host='127.0.0.1', port=8888)
    server.start()