CameraMachine/python/sockets/message_handler.py

from .connect_manager import ConnectionManager
from models import WebSocket
import json, asyncio
from mcu.DeviceControl import DeviceControl, checkMcuConnection
from mcu.BlueToothMode import BlueToothMode
from databases import DeviceConfig, SqlQuery, CRUD, PhotoRecord, SysConfigs
from mcu.capture.module_digicam import DigiCam
from mcu.capture.smart_shooter_class import SmartShooter
from models import RecordUpdate
from service.run_main import RunMain
from utils.utils_func import check_path
import time, shutil, os
from service.base import check_move_goods_art_no_folder
from service.deal_image import DealImage
import settings
from middleware import UnicornException
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from logger import logger
import stat
# 创建全局线程池
executor = ThreadPoolExecutor(max_workers=4)
async def handlerCutOut(
    manager=None, run_main=None, config_data={}, websocket=None, msg_type=""
):
    try:
        # return_data = run_main.check_before_cutout(config_data)
        # await run_main.check_for_cutout_image_first_call_back(return_data)
        # 将阻塞操作放到线程池中执行
        loop = asyncio.get_event_loop()
        return_data = await loop.run_in_executor(
            executor, partial(run_main.check_before_cutout, config_data)
        )
        # await run_main.check_for_cutout_image_first_call_back(return_data)
        await loop.run_in_executor(
            executor,
            partial(run_main.check_for_cutout_image_first_call_back, return_data),
        )
    except UnicornException as e:
        data = manager.jsonMessage(
            code=1,
            msg=e.msg,
            msg_type=msg_type,
        )
        await manager.send_personal_message(data, websocket)
        return
    except Exception as e:
        print("error",e)
        data = manager.jsonMessage(
            code=1,
            msg="抠图异常,请稍后重试~",
            msg_type=msg_type,
        )
        await manager.send_personal_message(data, websocket)

def handlerFolderDelete(limit_path, goods_art_no_arrays, is_write_txt_log):
    check_path(limit_path)
    move_folder_array = check_move_goods_art_no_folder(
        "output", goods_art_no_arrays, limit_path
    )
    
    for goods_art_revice in goods_art_no_arrays:
        cutout_goods = f"{limit_path}/{goods_art_revice}"
        if os.path.exists(cutout_goods):
            for root, dirs, files in os.walk(cutout_goods):
                for file in files:
                    filepath = os.path.join(root, file)
                    os.chmod(filepath, stat.S_IWRITE)
            logger.info(f"解除占用并开始删除目录:{cutout_goods}")
            # 尝试多次删除，增加成功率
            retry_count = 3
            while retry_count > 0:
                try:
                    shutil.rmtree(cutout_goods, onerror=settings.handle_remove_readonly)
                    del move_folder_array[goods_art_revice]
                    break
                except OSError as e:
                    logger.info(f"目录已经被删除-OSError:{str(e)}")
                    break
                except (PermissionError) as e:
                    retry_count -= 1
                    if retry_count == 0:
                        logger.info(f"抠图前目录删除出现问题-PermissionError:{str(e)};{goods_art_revice};{cutout_goods}")
                        if is_write_txt_log:
                            error_file_path = f"{cutout_goods}/异常说明-出现目录丢失或缺少图片请点开查看原因.txt"
                            with open(error_file_path, 'w', encoding='utf-8') as f:
                                f.write("目录删除失败\n")
                                f.write(f"原因: 文件被占用或没有删除权限\n")
                                f.write(f"错误信息: {str(e)}\n")
                                f.write(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                                f.write(f"请关闭可能正在使用此目录的程序后,为当前货号点击重拍后重试\n")
                        else:
                            raise UnicornException(f"目录检查出现问题:{str(e)},请关闭错误提示中的被占用文件")
                    else:
                        logger.info(f"抠图前目录删除出现问题--PermissionError:{str(e)};{retry_count}")
                        time.sleep(0.5)  # 等待0.5秒后重试
                except Exception as e:
                    retry_count -= 1
                    if retry_count == 0:
                        logger.info(f"抠图前目录删除出现问题--Exception:{str(e)};{goods_art_revice};{cutout_goods}")
                        if is_write_txt_log:
                            error_file_path = f"{cutout_goods}/异常说明-出现目录丢失或缺少图片请点开查看原因.txt"
                            with open(error_file_path, 'w', encoding='utf-8') as f:
                                f.write("目录删除失败\n")
                                f.write(f"原因: {str(e)}\n")
                                f.write(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                                f.write(f"请检查目录状态后,为当前货号点击重拍后重试\n")
                        else:
                            raise UnicornException(f"目录检查出现问题:{str(e)},请关闭错误提示中的被占用文件")
                    else:
                        logger.info(f"抠图前目录删除出现问题--Exception:{str(e)};{retry_count}")
                        time.sleep(0.5)  # 等待0.5秒后重试
    
    return move_folder_array
def validate_goods_art_no(goods_art_no):
        """
        验证货号输入是否包含特殊字符
        """
        import re
        # 定义不允许的特殊字符，主要是文件系统中可能导致路径问题的字符
        invalid_chars = r'[<>:"/\\|?*]'
        if re.search(invalid_chars, goods_art_no):
            # 找出所有非法字符
            invalid_found = re.findall(invalid_chars, goods_art_no)
            invalid_str = ', '.join(set(invalid_found))
            return False,f"货号包含非法字符: {invalid_str}，请修改后再提交"
        return True,""
# socket消息发送逻辑处理方法
async def handlerSend(
    manager: ConnectionManager,
    receiveData: str,
    websocket: WebSocket,
    smart_shooter: SmartShooter,
):
    loop = asyncio.get_event_loop()
    receiveData = json.loads(receiveData)
    # 处理消息发送逻辑
    receiveData = json.loads(receiveData.get("text"))
    print("receiveData", receiveData)
    jsonType = receiveData.get("type")
    code = receiveData.get("code")
    msg = receiveData.get("msg")
    data = receiveData.get("data")

    match jsonType:
        case "ping":
            """发送心跳"""
            data = manager.jsonMessage("pong")
            await manager.send_personal_message(data, websocket)
        case "pong":
            """发送心跳"""
            pass
        case "forward_message":
            data = receiveData.get("data")
            dictMsg = {"code": code, "msg": msg, "data": data}
            await manager.broadcast(dictMsg)
        case "connect_mcu":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            device_ctrl.mcu_exit = False
            # if device_ctrl.serial_ins.check_connect():
            #     print("未连接")
            loop.create_task(checkMcuConnection(device_ctrl), name="mcu")
        case "connect_bluetooth":
            blue_tooth = BlueToothMode(websocket_manager=manager)
            blue_tooth.bluetooth_exit = False
            # await  blue_tooth.main_func()
            print("连接蓝牙信息")
            loop.create_task(blue_tooth.main_func(), name="blue_tooth")
            # loop.close()
        case "init_mcu":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            # 是否强制初始化
            is_force_init = False
            loop.create_task(device_ctrl.initDevice(is_force_init), name="init_mcu")
        case "get_mcu_info":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            if not device_ctrl.is_running:
                device_ctrl.sendSocketMessage(
                    code=1, msg="mcu设备未连接,请先连接设备", device_status=0
                )
            device_ctrl.send_get_all_info_to_mcu()
        case "control_mcu":
            device_name = data.get("device_name")
            value = data.get("value")
            if (device_name == "" or device_name == None) or (
                value == "" or value == None
            ):
                data = manager.jsonMessage(code=1, msg="参数错误", msg_type="mcu")
                await manager.send_personal_message(data, websocket)
                return
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            await device_ctrl.controlDevice(device_name, value)
        case "stop_action":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            if device_ctrl.is_runn_action == True:
                print("动作执行中，停止")
                device_ctrl.is_stop_action = True
            else:
                print("动作没有执行，略过")
                data = manager.jsonMessage(code=0, msg="执行终止", msg_type="run_mcu_stop")
                await manager.send_personal_message(data, websocket)
                return
        case "run_mcu":
            msg_type = "run_mcu"
            action_info = data.get("action", "执行左脚程序")
            goods_art_no = data.get("goods_art_no", None)
            if goods_art_no == None or goods_art_no == "":
                # 判断货号是否存在
                data = manager.jsonMessage(
                    code=1, msg="goods_art_no不能为空", msg_type=msg_type
                )
                await manager.send_personal_message(data, websocket)
                return
            is_ok, msg = validate_goods_art_no(goods_art_no)
            if not is_ok:
                data = manager.jsonMessage(
                    code=1, msg=msg, msg_type=msg_type
                )
                await manager.send_personal_message(data, websocket)
                return
            session = SqlQuery()
            sys_configs = CRUD(SysConfigs)
            action_configs = sys_configs.read(
                session, conditions={"key": "action_configs"}
            )
            action_configs_json = json.loads(action_configs.value)
            action_flag = "left"
            if "右" in action_info:
                action_flag = "right"
            tab_id = action_configs_json.get(action_flag)
            photoRecord = CRUD(PhotoRecord)
            goods_art_record = photoRecord.read(
                session, conditions={"goods_art_no": goods_art_no,"delete_time": None}
            )
            if goods_art_record != None:
                data = manager.jsonMessage(
                    code=1,
                    msg=f"货号[{goods_art_no}]已存在,请勿重复拍摄~",
                    msg_type=msg_type,
                )
                await manager.send_personal_message(data, websocket)
                return
            crud = CRUD(DeviceConfig)
            condtions = {"tab_id": tab_id}
            all_devices = crud.read_all(
                session, conditions=condtions, order_by="action_index", ascending=True
            )
            if len(all_devices) == 0:
                # 判断是否有可用配置
                data = manager.jsonMessage(code=1, msg="当前没有可用配置")
                await manager.send_personal_message(data, websocket, msg_type=msg_type)
                return
            action_list = [dict(device.__dict__) for device in all_devices]
            print("执行拍摄 打印动作列表", condtions)
            # logger.info("执行拍摄 打印动作列表", json.dumps(action_list))
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.run_mcu_config(
                    action_list, goods_art_no, action_info, smart_shooter
                ),
                name="run_mcu_config",
            )
            session.close()
        case "run_mcu_single":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.run_mcu_config_single(
                    data, None, "run_mcu_single", -1, smart_shooter
                ),
                name="run_mcu_single",
            )
        case "handler_take_picture":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            await device_ctrl.controlDevice("laser_position", 0)
            blue_tooth = BlueToothMode(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            session = SqlQuery()
            crud = CRUD(PhotoRecord)
            record = crud.read(session=session, order_by="id", ascending=False,conditions={"delete_time": None})
            if record == None:
            # 发送失败消息
                data = manager.jsonMessage(
                            code=1,
                            msg="单拍失败,请先输入货号或扫码进行组合拍摄",
                            msg_type="handler_take_picture",
                        )
                await manager.send_personal_message(data, websocket)
                return
            try:
                limit_path = "{}/{}".format(settings.OUTPUT_DIR,
                    time.strftime("%Y-%m-%d", time.localtime(time.time()))
                )
                move_folder_array = handlerFolderDelete(limit_path,[record.goods_art_no],False)
            except UnicornException as e:
                data = manager.jsonMessage(
                        code=1,
                        msg=e.msg,
                        msg_type="handler_take_picture",
                    )
                await manager.send_personal_message(data, websocket)
                return
            loop.create_task(
                blue_tooth.remote_control_v2.handlerTakePhoto(smart_shooter,session,record),
                name="run_mcu_config",
            )
            await asyncio.sleep(2.5)
            await device_ctrl.controlDevice("laser_position", 1)
        case "re_take_picture":  # 重拍
            msg_type = "re_take_picture"
            record_id = data.get("record_id")
            session = SqlQuery()
            photoRecord = CRUD(PhotoRecord)
            goods_art_record = photoRecord.read(session, conditions={"id": record_id,"delete_time": None})
            if goods_art_record == None:
                data = manager.jsonMessage(
                    code=1,
                    msg=f"记录不存在，请核实后重新操作~",
                    msg_type=msg_type,
                )
                await manager.send_personal_message(data, websocket)
                return
            action_id = goods_art_record.action_id
            goods_art_no = goods_art_record.goods_art_no
            image_index = goods_art_record.image_index
            crud = CRUD(DeviceConfig)
            condtions = {"id": action_id}
            device_action = crud.read(session, conditions=condtions)
            result_dict = dict(device_action.__dict__)
            print("device_action打印输出====>>>", result_dict)
            if device_action == None:
                # 判断是否有可用配置
                data = manager.jsonMessage(code=1, msg="当前没有可用配置")
                await manager.send_personal_message(data, websocket, msg_type=msg_type)
                return
            # 清除图片记录,执行重拍
            reset_data = {"image_path": None}
            photoRecord.update(session, record_id, **reset_data)
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.run_mcu_config_single(
                    result_dict,
                    goods_art_no,
                    msg_type=msg_type,
                    image_index=image_index,
                    smart_shooter=smart_shooter,
                    action_id=record_id,
                ),
                name="run_mcu_config_single",
            )
            session.close()
        case "get_deviation":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.getDeviationInfo(),
                name="get_deviation",
            )
        case "set_deviation":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            value = data.get("value", None)
            action_name = data.get("action_name", None)
            loop.create_task(
                device_ctrl.set_deviation_cmd(value, action_name, "set"),
                name="set_deviation",
            )
        case "move_deviation":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            value = data.get("value", None)
            action_name = data.get("action_name", None)
            loop.create_task(
                device_ctrl.set_deviation_cmd(value, action_name, "move"),
                name="move_deviation",
            )
        case "get_mcu_other_info":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.getMcuOtherInfo(),
                name="mcu_other_set_get",
            )
        case "set_mcu_other_info":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.setMcuOtherInfo(data),
                name="setMcuOtherInfo",
            )
        case "send_command":
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            loop.create_task(
                device_ctrl.sendCommand(data.get("command", None)),
                name="sendCommand",
            )
        case "smart_shooter_getinfo":
            """
            获取相机信息,是否连接
            """
            loop.create_task(
                smart_shooter.GetCameraInfo(msg_type="smart_shooter_getinfo"),
                name="smart_shooter_getinfo",
            )
        case "smart_shooter_enable_preview":
            """
            启动相机或关闭实时预览
            """
            value = data.get("value", True)
            loop.create_task(
                smart_shooter.EnableCameraPreview(
                    enable_status=value, msg_type="smart_shooter_enable_preview"
                ),
                name="smart_shooter_enable_preview",
            )
        case "smart_shooter_get_camera_property":
            """
            启动相机或关闭实时预览
            """
            msg_type = "smart_shooter_get_camera_property"
            code = 0
            status, info = await smart_shooter.GetCameraProperty()
            code = 1 if status == False else 0
            msg = info if status == False else "操作成功"
            data = info if status == True else {}
            data = manager.jsonMessage(
                code=code,
                msg=msg,
                msg_type=msg_type,
                data=data,
            )
            await manager.send_personal_message(data, websocket)
        case "work_path":
            """
            启动相机或关闭实时预览
            """
            msg_type = "work_path"
            code = 0
            project_root = os.getcwd()
            work_path = f"{project_root}\\output"
            check_path(work_path)
            params = {"work_path": work_path}
            data = manager.jsonMessage(
                code=code,
                msg=msg,
                msg_type=msg_type,
                data=params,
            )
            await manager.send_personal_message(data, websocket)
        case "smart_shooter_photo_take":
            """
            获取相机信息,是否连接
            """
            device_ctrl = DeviceControl(
                websocket_manager=manager, smart_shooter=smart_shooter
            )
            await device_ctrl.controlDevice("laser_position", 0)
            # 兼容主图测试
            id = data.get("id", 0)
            goods_art_no = data.get("goods_art_no", "")
            if goods_art_no:
                try:
                    limit_path = "{}/{}".format(settings.OUTPUT_DIR,
                        time.strftime("%Y-%m-%d", time.localtime(time.time()))
                    )
                    move_folder_array = handlerFolderDelete(limit_path,[goods_art_no],False)
                except UnicornException as e:
                    data = manager.jsonMessage(
                            code=1,
                            msg=e.msg,
                            msg_type="smart_shooter_photo_take",
                        )
                    await manager.send_personal_message(data, websocket)
                    return
            is_af = True
            loop.create_task(
                smart_shooter.CameraShooter(
                    msg_type="smart_shooter_photo_take",
                    id=id,
                    goods_art_no=goods_art_no,
                    is_af=is_af,
                ),
                name="smart_shooter_photo_take",
            )
            await asyncio.sleep(2.5)
            await device_ctrl.controlDevice("laser_position", 1)
        case "digicam_take_picture":
            msg_type = "re_take_picture"
            id = data.get("id", 0)
            goods_art_no = data.get("goods_art_no", "")
            session = SqlQuery()
            photoRecord = CRUD(PhotoRecord)
            goods_art_record = photoRecord.read(session, conditions={"id": id,"delete_time": None})
            if goods_art_record == None:
                data = manager.jsonMessage(
                    code=1,
                    msg=f"记录不存在，请核实后重新操作~",
                    msg_type=msg_type,
                )
                await manager.send_personal_message(data, websocket)
                return
            reset_data = {"image_path": None}
            photoRecord.update(session, id, **reset_data)
            device_ctrl = DeviceControl(websocket_manager=manager)#
            loop.create_task(
                device_ctrl.only_take_photo(
                    goods_art_no=goods_art_no,
                    image_index=goods_art_record.image_index,
                    record_id=id,
                ),
                name="sendCommand",
            )
            session.close()
        case "segment_progress":
            msg_type = "segment_progress"
            obj = None
            token = data.get("token", "")
            token = "Bearer " + token
            uuid = data.get("uuid", "")
            run_main = RunMain(obj, token, uuid)
            goods_art_no_arrays = data.get("goods_art_no", [])
            limit_path = "{}/{}".format(settings.OUTPUT_DIR,
                time.strftime("%Y-%m-%d", time.localtime(time.time()))
            )
            try:
                move_folder_array = handlerFolderDelete(limit_path,goods_art_no_arrays,True)
            except UnicornException as e:
                data = manager.jsonMessage(
                        code=1,
                        msg=e.msg,
                        msg_type=msg_type,
                    )
                await manager.send_personal_message(data, websocket)
                return
            # 该数组表示是否需要后面的移动文件夹操作，减少重复抠图，提升抠图时间和速度
            session = SqlQuery()
            for goods_art_no in goods_art_no_arrays:
                pr = CRUD(PhotoRecord)
                images = pr.read_all(session, conditions={"goods_art_no": goods_art_no,"delete_time": None})
                if not images:
                    data = manager.jsonMessage(
                        code=1,
                        msg=f"商品货号【{goods_art_no}】在商品档案资料中不存在，请检查货号是否正确",
                        msg_type=msg_type,
                    )
                    await manager.send_personal_message(data, websocket)
                    return
                if move_folder_array.get(goods_art_no) == None:
                    image_dir = "{}/data/".format(os.getcwd()).replace("\\", "/")
                    check_path(image_dir)
                    for idx, itemImg in enumerate(images):
                        if itemImg.image_path == "" or itemImg.image_path == None:
                            data = manager.jsonMessage(
                                code=1,
                                msg=f"货号【{goods_art_no}】存在没有拍摄完成的图片,请重拍或删除后重试",
                                msg_type=msg_type,
                            )
                            await manager.send_personal_message(data, websocket)
                            return
                        new_file_name = (
                            str(idx)+"_"+str(itemImg.goods_art_no) + "_" + str(idx) + ".jpg"
                        )
                        if not os.path.exists(
                            image_dir + "/" + os.path.basename(new_file_name)
                        ):
                            shutil.copy(itemImg.image_path, image_dir + new_file_name)
                    dealImage = DealImage(image_dir)
                    resFlag, path = dealImage.dealMoveImage(
                        image_dir=image_dir,
                        callback_func=None,
                        goods_art_no=goods_art_no,
                    )
                    if not resFlag:
                        # path
                        data = manager.jsonMessage(
                            code=1,
                            msg=f"抠图操作异常,请检查目录是否存在，或者权限不足",
                            msg_type=msg_type,
                        )
                        await manager.send_personal_message(data, websocket)
                        return
            session.close()
            # try:
            cutOutMode = (
                "1"
                if settings.getSysConfigs("other_configs", "cutout_mode", "普通抠图")
                == "普通抠图"
                else "2"
            )
            config_data = {
                "image_dir": limit_path,
                "image_order": ("正面,侧视,背面,背侧,组合,组合2,组合3,组合4,组合5"),
                "goods_art_no": "",
                "goods_art_nos": goods_art_no_arrays,
                "is_check_number": False,
                "resize_image_view": "后跟",
                "cutout_mode": cutOutMode,
                "logo_path": "",
                "special_goods_art_no_folder_line": "",
                "is_use_excel": False,  # 是否使用excel
                "excel_path": "",  # excel路径
                "is_check_color_is_all": False,
                "cutout_is_pass": True,
                "assigned_page_dict": {},
                "detail_is_pass": True,
                "upload_is_pass": False,
                "upload_is_enable": settings.IS_UPLOAD_HLM,  # 是否上传到惠利玛商品库，通过config.ini得is_upload开启
                "is_filter": False,
                "temp_class": {},
                "temp_name": "",
                "temp_name_list": [],
                "target_error_folder": f"{limit_path}/软件-生成详情错误",
                "success_handler": [],
            }
            loop.create_task(
                handlerCutOut(
                    manager=manager,
                    run_main=run_main,
                    config_data=config_data,
                    websocket=websocket,
                    msg_type=msg_type,
                ),
                 name="handlerCutOut",
            )
        case _:
            data = manager.jsonMessage(code=1, msg="未知消息")
            await manager.send_personal_message(data, websocket)
            return