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 import settings, datetime import pandas as pd from utils.hlm_http_request import forward_request from utils.utils_func import check_path from sockets.socket_client import socket_manager from mcu.DeviceControl import DeviceControl import time, shutil, os from sqlalchemy import and_, asc, desc from functools import partial from service.deal_image import DealImage from databases import DeviceConfig, SysConfigs, SqlQuery, CRUD, select,DeviceConfigTabs from service.run_main import RunMain import importlib from service.auto_deal_pics.upload_pic import UploadPic from service.OnePicTest import OnePicTest import hashlib import win32api, win32gui, win32con from win32gui import EnumWindows, GetWindowText # from service.AutoDealPics import AutoDealPics # for plugin in settings.plugins: # module_path, class_name = plugin.rsplit(".", 1) # print("module_path", module_path, class_name) # module = importlib.import_module(module_path) # getattr(module, class_name) def calculate_md5(filepath): # 打开文件，以二进制只读模式打开 with open(filepath, "rb") as f: # 创建MD5哈希对象 md5hash = hashlib.md5() # 循环读取文件的内容并更新哈希对象 for chunk in iter(lambda: f.read(4096), b""): md5hash.update(chunk) # 返回MD5哈希的十六进制表示 return md5hash.hexdigest() @app.get("/") async def index(): # await socket_manager.send_message(msg="测试") return {"message": "Hello World"} @app.get("/scan_serials", description="扫描可用的设备端口") async def scanSerials(): """扫描串口""" ports = list_ports.comports() print("Scanning", ports) return {"message": "Hello World"} @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) def fromExcelHandler(params: HandlerDetail): excel_path = params.excel_path excel_df = pd.read_excel(excel_path, sheet_name=0, header=0) handler_result = [] handler_result_folder = "" if "文件夹名称" not in excel_df.columns: raise UnicornException("缺失 [文件夹名称] 列") if "商品货号" not in excel_df.columns: raise UnicornException("缺失 [商品货号] 列") for index, row in excel_df.iterrows(): goods_art_no_image_dir = str(row["文件夹名称"]) goods_art_no = str(row["商品货号"]) try: if not goods_art_no: raise UnicornException("货号不能为空") token = "Bearer " + params.token session = SqlQuery() pr = CRUD(PhotoRecord) images = pr.read_all(session, conditions={"goods_art_no": goods_art_no}) if not images: raise UnicornException("没有可用货号数据") image_dir = "{}/data/".format(os.getcwd()).replace("\\", "/") check_path(image_dir) for itemImg in images: if itemImg.image_path == "" or itemImg.image_path == None: raise UnicornException( f"货号【{goods_art_no}】存在没有拍摄完成的图片,请重拍或删除后重试" ) if not os.path.exists(image_dir + "/" + os.path.basename(itemImg.image_path)): shutil.copy(itemImg.image_path, image_dir) dealImage = DealImage(image_dir) resFlag, path = dealImage.dealMoveImage( image_dir=image_dir, callback_func=None, goods_art_no=goods_art_no_image_dir, ) if not resFlag: raise UnicornException(path) temp_class = {} temp_name_list = [] for tempItem in params.temp_list: temp_class[tempItem.template_id] = tempItem.template_local_classes temp_name_list.append(tempItem.template_id) # if goods_art_no_image_dir not in path: # path = path + "/" + goods_art_no_image_dir config_data = { "image_dir": path, "image_order": params.template_image_order, "goods_art_no": goods_art_no, "is_check_number": False, "resize_image_view": "后跟", "cutout_mode": settings.CUTOUT_MODE, "logo_path": params.logo_path, "special_goods_art_no_folder_line": "", "is_use_excel": ( False if params.excel_path == "" else True ), # 是否使用excel "excel_path": params.excel_path, # excel路径 "is_check_color_is_all": False, "cutout_is_pass": False, "assigned_page_dict": {}, "detail_is_pass": False, "upload_is_pass": False, "upload_is_enable": False, "is_filter": False, "temp_class": temp_class, "temp_name": params.temp_name, "temp_name_list": temp_name_list, "target_error_folder": f"{path}/软件-生成详情错误", } # 动态导入类 temp_class_dict = {} for key, class_path in config_data["temp_class"].items(): module_path, class_name = class_path.rsplit(".", 1) module = importlib.import_module(module_path) cls = getattr(module, class_name) temp_class_dict[key] = cls config_data["temp_class"] = temp_class_dict obj = None run_main = RunMain(obj, token) return_data = run_main.check_before_cutout(config_data) cutout_res = run_main.check_for_cutout_image_first_call_back(return_data) check_for_detail_first_res = None if cutout_res == True: return_data_check_before_detail = run_main.check_before_detail(config_data) print( "return_data_check_before_detail======> 测试 ==>", return_data_check_before_detail, ) check_for_detail_first_res = run_main.check_for_detail_first_call_back( return_data_check_before_detail ) if isinstance(check_for_detail_first_res, partial): result = check_for_detail_first_res() try: config_data = result["config_data"] except: config_data = result if config_data["sign_text"] == "已结束详情处理": # at_pic = AutoDealPics() print("config_data", config_data) if config_data["upload_is_enable"]: to_deal_dir = "{}/软件-详情图生成".format(config_data["image_dir"]) check_path(to_deal_dir) print("to_deal_dir", to_deal_dir) if os.path.exists(to_deal_dir): upload_pic = UploadPic( windows=None, to_deal_dir=to_deal_dir, config_data=config_data, token=token, ) upload_pic.run() out_put_dir = config_data["out_put_dir"] out_put_dir_path = "{}/{}".format(os.getcwd(), out_put_dir).replace("\\", "/") handler_result_folder = os.path.dirname(out_put_dir_path) handler_result.append({"goods_art_no": goods_art_no, "success": True, "info": "处理成功"}) else: handler_result.append({"goods_art_no": goods_art_no, "success": False, "info": "处理失败"}) except Exception as e: handler_result.append({"goods_art_no": goods_art_no, "success": False, "info": str(e)}) handler_result_folder = "/".join(handler_result_folder.split("/")[:-1]) return { "code": 0, "msg": "", "data": {"output_folder": handler_result_folder, "list": handler_result}, } @app.post("/handle_detail") async def handle_detail(request: Request, params: HandlerDetail): goods_art_no_array = params.goods_art_no is_only_cutout = params.is_only_cutout handler_result = [] handler_result_folder = "" if params.excel_path != "" and params.excel_path != None: return fromExcelHandler(params) for goods_art_no in goods_art_no_array: try: if not goods_art_no: raise UnicornException("货号不能为空") token = "Bearer " + params.token session = SqlQuery() pr = CRUD(PhotoRecord) images = pr.read_all(session, conditions={"goods_art_no": goods_art_no}) if not images: raise UnicornException("没有可用货号数据") detail_counts = len(params.template_image_order.split(",")) image_counts = len(images) if image_counts < detail_counts: raise UnicornException( f"货号:[{goods_art_no}],实际照片数量:{image_counts}张,小于详情图要求数量:{detail_counts}张" ) image_dir = "{}/data/".format(os.getcwd()).replace("\\", "/") check_path(image_dir) for itemImg in images: if itemImg.image_path == "" or itemImg.image_path == None: raise UnicornException( f"货号【{goods_art_no}】存在没有拍摄完成的图片,请重拍或删除后重试" ) if not os.path.exists(image_dir + "/" + os.path.basename(itemImg.image_path)): shutil.copy(itemImg.image_path, image_dir) dealImage = DealImage(image_dir) resFlag, path = dealImage.dealMoveImage( image_dir=image_dir, callback_func=None, goods_art_no=goods_art_no ) if not resFlag: raise UnicornException(path) temp_class = {} temp_name_list = [] for tempItem in params.temp_list: temp_class[tempItem.template_id] = tempItem.template_local_classes temp_name_list.append(tempItem.template_id) config_data = { "image_dir": path, "image_order": params.template_image_order, "goods_art_no": goods_art_no, "is_check_number": False, "resize_image_view": "后跟", "cutout_mode": settings.CUTOUT_MODE, "logo_path": params.logo_path, "special_goods_art_no_folder_line": "", "is_use_excel": ( False if params.excel_path == "" else True ), # 是否使用excel "excel_path": params.excel_path, # excel路径 "is_check_color_is_all": False, "cutout_is_pass": False, "assigned_page_dict": {}, "detail_is_pass": False, "upload_is_pass": False, "upload_is_enable": False, "is_filter": False, "temp_class": temp_class, "temp_name": params.temp_name, "temp_name_list": temp_name_list, "target_error_folder": f"{path}/软件-生成详情错误", } # 动态导入类 temp_class_dict = {} for key, class_path in config_data["temp_class"].items(): module_path, class_name = class_path.rsplit(".", 1) module = importlib.import_module(module_path) cls = getattr(module, class_name) temp_class_dict[key] = cls config_data["temp_class"] = temp_class_dict obj = None run_main = RunMain(obj, token) return_data = run_main.check_before_cutout(config_data) cutout_res = run_main.check_for_cutout_image_first_call_back(return_data) check_for_detail_first_res = None if is_only_cutout == 1 and cutout_res == True: out_put_dir = return_data["data"]["image_dir"] + "/" + goods_art_no out_put_dir_path = "{}/{}".format(os.getcwd(), out_put_dir).replace( "\\", "/" ) # print("out_put_dir_path", out_put_dir_path) handler_result_folder = os.path.dirname(out_put_dir_path) handler_result.append( {"goods_art_no": goods_art_no, "success": True, "info": "处理成功"} ) # print("检测到需要裁剪图片，开始裁剪图片", return_data) continue if cutout_res == True: return_data_check_before_detail = run_main.check_before_detail(config_data) print( "return_data_check_before_detail======> 测试 ==>", return_data_check_before_detail, ) check_for_detail_first_res = run_main.check_for_detail_first_call_back( return_data_check_before_detail ) if isinstance(check_for_detail_first_res, partial): result = check_for_detail_first_res() try: config_data = result["config_data"] except: config_data = result if config_data["sign_text"] == "已结束详情处理": # at_pic = AutoDealPics() print("config_data", config_data) if config_data["upload_is_enable"]: to_deal_dir = "{}/软件-详情图生成".format(config_data["image_dir"]) check_path(to_deal_dir) print("to_deal_dir", to_deal_dir) if os.path.exists(to_deal_dir): upload_pic = UploadPic( windows=None, to_deal_dir=to_deal_dir, config_data=config_data, token=token, ) upload_pic.run() out_put_dir = config_data["out_put_dir"] out_put_dir_path = "{}/{}".format(os.getcwd(), out_put_dir).replace("\\", "/") handler_result_folder = os.path.dirname(out_put_dir_path) handler_result.append({"goods_art_no": goods_art_no, "success": True, "info": "处理成功"}) else: handler_result.append({"goods_art_no": goods_art_no, "success": False, "info": "处理失败"}) except Exception as e: handler_result.append({"goods_art_no": goods_art_no, "success": False, "info": str(e)}) return { "code": 0, "msg": "", "data": {"output_folder": handler_result_folder, "list": handler_result}, } @app.get("/get_device_tabs", description="获取可执行程序命令列表") def get_device_tabs(type:int): session = SqlQuery() statement = ( select(DeviceConfigTabs) .where(DeviceConfigTabs.mode_type == type).order_by(asc("id")) ) result = session.exec(statement).all() session.close() sys = CRUD(SysConfigs) action_configs = sys.read(session, conditions={"key": "action_configs"}) session.close() return { "code": 0, "msg": "", "data": {"tabs": result, "select_configs": json.loads(action_configs.value)}, } @app.post("/update_tab_name", description="更改tab名称") def update_tab_name(params: DeviceConfigTabsReq): if params.mode_name == "": return {"code": 1, "msg": "名称不能为空", "data": {}} session = SqlQuery() tabModel = CRUD(DeviceConfigTabs) kwargs = {"mode_name": params.mode_name} tabModel.updateConditions(session, conditions={"id": params.id}, **kwargs) session.close() return { "code": 0, "msg": "", "data": None, } @app.post("/get_device_configs", description="获取可执行程序命令列表") def get_device_configs(params: ModelGetDeviceConfig): tab_id = params.tab_id session = SqlQuery() configModel = CRUD(DeviceConfig) configList = configModel.read_all( session, conditions={"tab_id": tab_id}, 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("/device_config_detail_query", description="通过条件获取可执行程序详情") def device_config_detail_query(): # tab_id = params.tab_id # action_name = params.action_name session = SqlQuery() sys = CRUD(SysConfigs) action_configs = sys.read(session, conditions={"key": "action_configs"}) action_configs_value = json.loads(action_configs.value) left_config = action_configs_value.get("left") configModel = CRUD(DeviceConfig) model = configModel.read( session, conditions={"tab_id": left_config, "action_name": "侧视"} ) if model == None: model = configModel.read( session, conditions={"tab_id": left_config} ) 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} if model.is_system == True: return {"code": 1, "msg": "系统配置不允许删除", "data": None} configArray = configModel.read_all(session, conditions={"tab_id": model.tab_id}) if len(configArray) == 1: 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): tab_id = params.tab_id if tab_id == None or tab_id == "": return {"code": 1, "msg": "参数错误", "data": None} session = SqlQuery() deviceConfig = CRUD(DeviceConfig) first_config = deviceConfig.read(session, conditions={"tab_id": tab_id}) res = deviceConfig.deleteConditions(session, conditions={"tab_id": tab_id}) if res is False: return {"code": 1, "msg": "操作失败", "data": None} actions = json.load(open("action.json", encoding="utf-8")) for data in actions: data["tab_id"] = tab_id data["is_system"] = first_config.is_system device_config = DeviceConfig(**data) session.add(device_config) session.commit() session.close() return {"code": 0, "msg": "操作成功", "data": None} @app.get("/get_photo_records", description="获取拍照记录") def get_photo_records(page: int = 1, size: int = 5): session = SqlQuery() photos = CRUD(PhotoRecord) print("准备查询拍摄记录", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) statement = ( select(PhotoRecord) .offset((page - 1) * size) .limit(size) .order_by(desc("id")) .group_by("goods_art_no") ) list = [] result = session.exec(statement).all() print("group 完成 ", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) join_conditions = [ { "model": DeviceConfig, "on": PhotoRecord.action_id == DeviceConfig.id, "is_outer": False, # 可选，默认False，设为True则为LEFT JOIN } ] for item in result: print(item) # list_item = photos.read_all( # session, # conditions={"goods_art_no": item.goods_art_no}, # join_conditions=join_conditions, # ) query = ( select(PhotoRecord, DeviceConfig.action_name) .where(PhotoRecord.goods_art_no == item.goods_art_no) .join(DeviceConfig, PhotoRecord.action_id == DeviceConfig.id) ) list_item = session.exec(query).mappings().all() session.close() list.append( { "goods_art_no": item.goods_art_no, "action_time": item.create_time, "items": list_item, } ) session.close() print("循环查询 完成 ", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) return { "code": 0, "msg": "", "data": {"list": list, "page": page, "size": size}, } @app.get("/get_last_photo_record", description="获取最后一条拍照记录") def get_last_photo_record(): session = SqlQuery() statement = select(PhotoRecord).where(PhotoRecord.image_path != None).order_by(desc("photo_create_time")) result = session.exec(statement).first() session.close() return { "code": 0, "msg": "", "data": result, } @app.get("/get_photo_record_detail", description="通过货号获取拍照记录详情") def get_photo_records(goods_art_no: str = None): if goods_art_no == None: return {"code": 1, "msg": "参数错误", "data": None} session = SqlQuery() photos = CRUD(PhotoRecord) items = photos.read_all(session, conditions={"goods_art_no": goods_art_no}) session.close() return { "code": 0, "msg": "", "data": {"list": items}, } @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}) session.close() return { "code": 0, "msg": "操作成功", "data": None, } @app.get("/get_sys_config", description="查询系统配置") def get_sys_config(key: str = None): if key == None: return {"code": 1, "msg": "参数错误", "data": None} session = SqlQuery() photos = CRUD(SysConfigs) item = photos.read(session, conditions={"key": key}) session.close() return { "code": 0, "msg": "", "data": json.loads(item.value), } @app.post("/update_left_right_config", description="更新左右脚配置") def update_left_right_config(params: LeftRightParams): session = SqlQuery() sysConfig = CRUD(SysConfigs) model = sysConfig.read(session, conditions={"key": "action_configs"}) if model == None: return {"code": 1, "msg": "配置不存在", "data": None} config_value = json.loads(model.value) config_value[params.type] = params.id update_value = json.dumps(config_value) # 走编辑逻辑 kwargs = {"key": "action_configs", "value": update_value} save_device_config = sysConfig.updateConditions( session, conditions={"key": "action_configs"}, **kwargs ) # return {"code": 0, "msg": "操作成功", "data": save_device_config} return {"code": 0, "msg": "操作成功", "data": None} @app.post("/update_sys_configs", description="创建或修改系统配置") def save_sys_configs(params: SysConfigParams): session = SqlQuery() sysConfig = CRUD(SysConfigs) model = sysConfig.read(session, conditions={"key": params.key}) if model == None: return {"code": 1, "msg": "配置不存在", "data": None} # 走编辑逻辑 kwargs = params.__dict__ save_device_config = sysConfig.updateConditions(session, conditions={"key": params.key}, **kwargs) return {"code": 0, "msg": "操作成功", "data": save_device_config} @app.post("/create_main_image", description="创建主图测试") def create_main_image(params: MaineImageTest): file_path = params.file_path onePic = OnePicTest(pic_path=file_path) # session = SqlQuery() # sysConfig = CRUD(SysConfigs) # model = sysConfig.read(session, conditions={"key": params.key}) # if model == None: # return {"code": 1, "msg": "配置不存在", "data": None} # # 走编辑逻辑 # kwargs = params.__dict__ # save_device_config = sysConfig.updateConditions( # session, conditions={"key": params.key}, **kwargs # ) main_out_path = onePic.HandlerMainImage() return {"code": 0, "msg": "操作成功", "data": {"main_out_path": main_out_path}} def insertEmptyLogoList(session): """插入空logo列表""" data = {"key": "logo_configs", "value": "[]"} config = SysConfigs(**data) session.add(config) session.commit() session.close() item = SysConfigs() item.key = "logo_configs" item.value = "[]" return item @app.get("/logo_list", description="logo列表") def logo_list(): logo_dir = "{}/data/logo/".format(os.getcwd()).replace("\\", "/") check_path(logo_dir) logo_files = os.listdir(logo_dir) logo_list = [] for logoItem in logo_files: logo_list.append(f"{logo_dir}{logoItem}") return {"code": 0, "msg": "操作成功", "data": logo_list} @app.post("/add_logo", description="添加logo") def add_logo(params: LogoParams): logo_path = params.logo_path.replace("\\", "/") session = SqlQuery() sysConfig = CRUD(SysConfigs) item = sysConfig.read(session, conditions={"key": "logo_configs"}) if item == None: item = insertEmptyLogoList(session) if os.path.isfile(logo_path) == False: return {"code": 1, "msg": "logo文件不存在", "data": None} logo_dir = "{}/data/logo/".format(os.getcwd()).replace("\\", "/") check_path(logo_dir) fpath, fname = os.path.split(logo_path) logo_path_info = logo_dir + fname shutil.copy(logo_path, logo_path_info) # 复制文件 logo_files = os.listdir(logo_dir) logo_list = [] for logoItem in logo_files: logo_list.append(f"{logo_dir}{logoItem}") return { "code": 0, "msg": "", "data": {"logo": logo_path_info}, } @app.post("/delete_logo", description="删除logo") def delete_logo(params: LogoParamsupdate): logo_path = params.path if os.path.isfile(logo_path) == False: return {"code": 1, "msg": "logo文件不存在", "data": None} os.remove(logo_path) logo_dir = "{}/data/logo/".format(os.getcwd()).replace("\\", "/") check_path(logo_dir) logo_files = os.listdir(logo_dir) logo_list = [] for logoItem in logo_files: logo_list.append(f"{logo_dir}{logoItem}") return {"code": 0, "msg": "操作成功", "data": logo_list} @app.post("/close_other_window", description="关闭窗口") def close_other_window(): hwnd_list = [] def callback(hwnd, _): title = GetWindowText(hwnd) if title == "digiCamControl by Duka Istvan": hwnd_list.append(hwnd) EnumWindows(callback, None) if hwnd_list: hwnd = hwnd_list[0] win32gui.PostMessage(hwnd, win32con.WM_CLOSE, 0, 0) return {"code": 0, "msg": "关闭成功", "data": {"status":True}} return {"code": 0, "msg": "关闭失败", "data": {"status": False}}