from natsort.natsort import order_by_index from sqlalchemy import func from models import * import requests import json from logger import logger from serial.tools import list_ports from model import PhotoRecord 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, select @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': , 'huilima-3': , 'huilima-4': , 'huilima-1': }, '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} @app.get("/get_photo_records", description="获取拍照记录") def get_photo_records(page: int=1,size:int=3): session = SqlQuery() photos = CRUD(PhotoRecord) statement = ( select(PhotoRecord) .offset((page - 1) * size) .limit(size) .group_by("goods_art_no") ) list = [] result = session.exec(statement).all() for item in result: list_item = photos.read_all(session, conditions={"goods_art_no": item.goods_art_no}) list.append({"goods_art_no": item.goods_art_no, "items": list_item}) return { "code": 0, "msg": "", "data": {"list": list, "page": page, "size": size}, } @app.post("/delect_goods_arts", description="通过货号删除记录") def delect_goods_arts(params: PhotoRecordDelete): session = SqlQuery() photos = CRUD(PhotoRecord) for item in params.goods_art_nos: photos.deleteConditions(session, conditions={"goods_art_no": item}) return { "code": 0, "msg": "操作成功", "data": None, }