CameraMachine/python/api.py

from natsort.natsort import order_by_index
from models import *
import requests
import json
from logger import logger
from serial.tools import list_ports
from utils.hlm_http_request import forward_request
from sockets.socket_client import socket_manager
from mcu.DeviceControl import DeviceControl
import time
# from service.base_deal import  BaseDealImage
from databases import DeviceConfig,SqlQuery,CRUD

@app.get("/")
async def index():
    # await socket_manager.send_message(msg="测试")
    return {"message": "Hello World"}


@app.get("/send_test")
async def index():
    data = {"data1":1,"data2":2,"data3":3,"data4":4}
    await socket_manager.send_message(msg="测试",data=data)
    return {"message": "Hello World"}


@app.get("/scan_serials", description="扫描可用的设备端口")
async def scanSerials():
    """扫描串口"""
    ports = list_ports.comports()
    print("Scanning", ports)
    return {"message": "Hello World"}


@app.get("/test_conndevice")
def test_conndevice():
        device_control = DeviceControl()
        p_list = []
        temp_ports_dict = {}
    # while 1:
        time.sleep(1)
        ports_dict = device_control.scan_serial_port()
        temp_ports_dict = ports_dict

        if not ports_dict:
            # 全部清空 移除所有串口
            if p_list:
                _p = p_list.pop()
                device_control.remove_port(_p)
            # continue

        if ports_dict:
            # print(plist)
            for index, _i in enumerate(p_list):
                if _i not in ports_dict:
                    _p = p_list.pop(index)
                    device_control.remove_port(_p)

            for _port_name, _port_value in ports_dict.items():
                if _port_name not in p_list:
                    try:
                        p_list.append(_port_name)
                        device_control.add_port_by_linkage(_port_name)
                    except BaseException as e:
                        print(e.__traceback__.tb_frame.f_globals["__file__"])   # 发生异常所在的文件
                        print(e.__traceback__.tb_lineno)                        # 发生异常所在的行数
                        print("串口不存在{} {}".format(_port_name, e))

                # threading.Thread(target=self.add_port, args=(_port_name, _port_value)).start()
                # self.add_port(_p)


@app.api_route(
    "/forward_request", methods=["GET", "POST"], description="代理转发hlm项目得请求"
)
async def forwardRequest(request: HlmForwardRequest):
    """
    转发HTTP请求到目标URL

    :param request: FastAPI Request对象
    :return: 目标接口的响应
    """
    try:
        if request.method == "GET":
            params = request.query_params
        elif request.method == "POST":
            params = json.dump(request.query_params)
        else:
            raise UnicornException("仅支持GET和POST方法")
        target_url = request.target_url
        method = request.method.upper()
        headers = request.headers
        if not target_url:
            raise UnicornException("目标url地址不能为空")
        # 调用 hlm_http_request 中的 forward_request 函数
        response = forward_request(
            target_url, params=params, method=method, headers=headers
        )
        return response
    except requests.RequestException as e:
        raise UnicornException(e)
    except Exception as e:
        raise UnicornException(e)

    # @app.post('/handle_detail')
    # async def handle_detail(request: Request):
    #
    #     image_dir = "{}/data".format(os.getcwd())
    #     baseDealImage = BaseDealImage(image_dir=image_dir)
    #     result = baseDealImage.dealMoveImage(image_dir=image_dir, callback_func=None)
    #

    #
    # params = json.dump(request.query_params)
    # #{'image_dir': 'D:/phpstudy_pro/WWW/auto_photo/output/2024-11-18', 'image_order': '俯视,侧视,后跟,鞋底,内里', 'is_check_number': True, 'resize_image_view': '后跟', 'cutout_mode': '1', 'logo_path': '', 'special_goods_art_no_folder_line': '', 'is_use_excel': True, 'excel_path': '', 'is_check_color_is_all': True, 'assigned_page_dict': {}, 'temp_class': {'huilima-2': <class 'detail_template.huilima.detail_huilima2.DetailPicGet'>, 'huilima-3': <class 'detail_template.huilima.detail_huilima3.DetailPicGet'>, 'huilima-4': <class 'detail_template.huilima.detail_huilima4.DetailPicGet'>, 'huilima-1': <class 'detail_template.huilima.detail_huilima1.DetailPicGet'>}, 'temp_name': 'huilima-2', 'temp_name_list': ['huilima-2', 'huilima-3', 'huilima-4', 'huilima-1'], 'target_error_folder': 'D:/phpstudy_pro/WWW/auto_photo/output/2024-11-18/软件-生成详情错误'}
    #
    # config_data = {
    #     'image_dir': params['image_dir'],
    #     'image_order': params['image_order'],
    #     'is_check_number': params['is_check_number'],
    #     'resize_image_view': params['resize_image_view'],
    #     'cutout_mode': '1',
    #     'logo_path': params['logo_path'],
    #     'special_goods_art_no_folder_line': '',
    #     'is_use_excel': params['is_use_excel'],
    #     'excel_path': params['excel_path'],
    #     'is_check_color_is_all': params['is_check_color_is_all'],
    #     'assigned_page_dict': {},
    #     'temp_class': {
    #         'huilima-2': 'detail_template.huilima.detail_huilima2.DetailPicGet',
    #         'huilima-3': 'detail_template.huilima.detail_huilima3.DetailPicGet',
    #         'huilima-4': 'detail_template.huilima.detail_huilima4.DetailPicGet',
    #         'huilima-1': 'detail_template.huilima.detail_huilima1.DetailPicGet'
    #     },
    #     'temp_name': 'huilima-2',
    #     'temp_name_list': ['huilima-2', 'huilima-3', 'huilima-4', 'huilima-1'],
    #     'target_error_folder': 'D:/phpstudy_pro/WWW/auto_photo/output/2024-11-18/软件-生成详情错误'
    # }

@app.post("/get_device_configs", description="获取可执行程序命令列表")
def get_device_configs(params: ModelGetDeviceConfig):
    mode_type = params.mode_type
    session = SqlQuery()
    configModel = CRUD(DeviceConfig)
    configList = configModel.read_all(
        session, conditions={"mode_type": mode_type}, order_by="action_index", ascending=True
    )
    return {"code": 0, "msg": "", "data": {"list": configList}}


@app.post("/device_config_detail", description="获取可执行程序详情")
def get_device_configs(params: ModelGetDeviceConfigDetail):
    action_id = params.id
    session = SqlQuery()
    configModel = CRUD(DeviceConfig)
    model = configModel.read(session, conditions={"id": action_id})
    if model == None:
        return {"code": 1, "msg": "数据不存在", "data": None}
    return {"code": 0, "msg": "", "data": model}


@app.post("/remove_config", description="删除一条可执行命令")
def get_device_configs(params: ModelGetDeviceConfigDetail):
    action_id = params.id
    session = SqlQuery()
    configModel = CRUD(DeviceConfig)
    model = configModel.read(session, conditions={"id": action_id})
    if model == None:
        return {"code": 1, "msg": "数据不存在", "data": None}
    configModel.delete(session, obj_id=action_id)
    return {"code": 0, "msg": "删除成功", "data": None}


@app.post("/save_device_config", description="创建或修改一条可执行命令")
def save_device_config(params: SaveDeviceConfig):
    action_id = params.id
    session = SqlQuery()
    deviceConfig = CRUD(DeviceConfig)
    if action_id == None or action_id == 0:
        # 走新增逻辑
        params.id = None
        save_device_config = deviceConfig.create(session, obj_in=params)
    else:
        model = deviceConfig.read(session, conditions={"id": action_id})
        if model == None:
            return {"code": 1, "msg": "数据不存在", "data": None}
        # 走编辑逻辑
        kwargs = params.__dict__
        save_device_config = deviceConfig.update(session, obj_id=action_id, **kwargs)
    return {"code": 0, "msg": "操作成功", "data": save_device_config}


@app.post("/reset_config", description="创建或修改一条可执行命令")
def reset_config(params: ModelGetDeviceConfig):
    mode_type = params.mode_type
    if  mode_type == None or mode_type == "":
        return {"code": 1, "msg": "参数错误", "data": None}
    session = SqlQuery()
    deviceConfig = CRUD(DeviceConfig)
    res = deviceConfig.deleteConditions(session, conditions={"mode_type": mode_type})
    if res is False:
        return {"code": 1, "msg": "操作失败", "data": None}
    actions = json.load(open("action.json", encoding="utf-8"))
    act = []
    for item in actions:
        if item.get("mode_type") == mode_type:
            act.append(item)
    batch_insert_device_configs(session, act)
    return {"code": 0, "msg": "操作成功", "data": None}