CameraMachine/python/api.py

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("没有可用货号数据")
            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 = 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()
    sys = CRUD(SysConfigs)
    action_configs = sys.read(session, conditions={"key": "action_configs"})
    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
    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")
    session = SqlQuery()
    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}
    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)
    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"))
    act = []
    for item in actions:
        if item.get("tab_id") == tab_id:
            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 = 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"))
    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,
                "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_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}}