tuoheng_alg_move_20240715/util/ImgBaiduSdk.py

# -*- coding: utf-8 -*-
import time
import traceback
from os.path import join

from aip import AipImageClassify, AipBodyAnalysis
from loguru import logger

from common.YmlConstant import baidu_yml_path
from enums.BaiduSdkEnum import BAIDUERRORDATA
from enums.ExceptionEnum import ExceptionType
from exception.CustomerException import ServiceException
from util.ImageUtils import url2Content
from util.RWUtils import getConfigs


class AipImageClassifyClient:

    __slots__ = ('__client', '__config')

    def __init__(self, base_dir, env):
        self.__client = None
        self.__config = getConfigs(join(base_dir, baidu_yml_path % env))
        self.init_client()
        # self.lock = Lock()

    def init_client(self):
        if self.__client is None:
            self.__client = AipImageClassify(str(self.__config["vehicle"]["APP_ID"]),
                                             self.__config["vehicle"]["API_KEY"],
                                             self.__config["vehicle"]["SECRET_KEY"])

    '''
        车辆检测
    '''

    def vehicleDetectUrl(self, url, request_id, options={}):
        self.init_client()
        # try:
        #     self.lock.acquire()
        reply_num = 0
        reply_value = None
        while True:
            try:
                options["show"] = "true"
                res_image = self.__client.vehicleDetectUrl(url, options)
                error_code = res_image.get("error_code")
                if error_code:
                    enum = BAIDUERRORDATA.get(error_code)
                    # 如果异常编码未知, 返回空值
                    if enum is None:
                        logger.error("百度云车辆检测异常！error_code:{}, request_id: {}", error_code, request_id)
                        return None
                    # 重试指定次数后，还是异常，输出统一内部异常
                    if enum.value[3] == 0:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        logger.error("百度云车辆检测异常！error_code:{}, error_msg:{}, reply_num:{}, request_id: {}",
                                     enum.value[0], enum.value[2], reply_num, request_id)
                        raise Exception()
                    # 重试指定次数后，还是异常，输出对应的异常
                    if enum.value[3] == 1:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        raise ServiceException(str(enum.value[0]), enum.value[2])
                    # 重试指定次数后，还是异常，输出空
                    if enum.value[3] == 2:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        if reply_num >= reply_value:
                            return None
                        raise Exception()
                return res_image
            except Exception as e:
                time.sleep(1)
                reply_num += 1
                self.init_client()
                if reply_num > reply_value:
                    if isinstance(e, ServiceException):
                        raise ServiceException(e.code, e.msg)
                    logger.error("车辆检测识别失败: {}, request_id: {}", traceback.format_exc(), request_id)
                    raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                           ExceptionType.SERVICE_INNER_EXCEPTION.value[1])
        # except Exception as ee:
        #     logger.exception("车辆检测加锁异常: {}, request_id: {}", ee, request_id)
        #     raise ServiceException(ExceptionType.UNIVERSAL_TEXT_RECOGNITION_FAILED.value[0],
        #                            ExceptionType.UNIVERSAL_TEXT_RECOGNITION_FAILED.value[1])
        # finally:
        #     self.lock.release()

    '''
        车辆检测
    '''

    def vehicleDetect(self, iamge, request_id, options={}):
        self.init_client()
        # try:
        #     self.lock.acquire()
        reply_num = 0
        reply_value = None
        while True:
            try:
                options["show"] = "true"
                res_image = self.__client.vehicleDetect(iamge, options)
                error_code = res_image.get("error_code")
                if error_code:
                    enum = BAIDUERRORDATA.get(error_code)
                    # 如果异常编码未知, 返回空值
                    if enum is None:
                        logger.error("百度云车辆检测异常！error_code:{}, request_id: {}", error_code, request_id)
                        return None
                    # 重试指定次数后，还是异常，输出统一内部异常
                    if enum.value[3] == 0:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        logger.error("百度云车辆检测异常！error_code:{}, error_msg:{}, reply_num:{}, request_id: {}",
                                     enum.value[0], enum.value[2], reply_num, request_id)
                        raise Exception()
                    # 重试指定次数后，还是异常，输出对应的异常
                    if enum.value[3] == 1:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        raise ServiceException(str(enum.value[0]), enum.value[2])
                    # 重试指定次数后，还是异常，输出空
                    if enum.value[3] == 2:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        if reply_num >= reply_value:
                            return None
                        raise Exception()
                return res_image
            except Exception as e:
                time.sleep(1)
                reply_num += 1
                self.init_client()
                if reply_num > reply_value:
                    if isinstance(e, ServiceException):
                        raise ServiceException(e.code, e.msg)
                    logger.error("车辆检测识别失败: {}, request_id: {}", traceback.format_exc(), request_id)
                    raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                           ExceptionType.SERVICE_INNER_EXCEPTION.value[1])


class AipBodyAnalysisClient:

    __slots__ = ('__config', '__client')

    def __init__(self, base_dir, env):
        self.__client = None
        self.__config = getConfigs(join(base_dir, baidu_yml_path % env))
        self.init_client()
        # self.lock = Lock()

    def init_client(self):
        if self.__client is None:
            self.__client = AipBodyAnalysis(str(self.__config["person"]["APP_ID"]), self.__config["person"]["API_KEY"],
                                            self.__config["person"]["SECRET_KEY"])

    '''
        人体检测与属性识别
    '''

    def bodyAttr(self, url, request_id, options={}):
        self.init_client()
        image = self.readImage(url, request_id)
        # try:
        #     self.lock.acquire()
        reply_num = 0
        reply_value = None
        while True:
            try:
                options["show"] = "true"
                res_image = self.__client.bodyAttr(image, options)
                error_code = res_image.get("error_code")
                if error_code:
                    enum = BAIDUERRORDATA.get(error_code)
                    # 如果异常编码未知, 返回空值
                    if enum is None:
                        logger.error("百度云人体检测与属性识别异常！error_code:{}, request_id: {}", error_code, request_id)
                        return None
                    # 重试指定次数后，还是异常，输出统一内部异常
                    if enum.value[3] == 0:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        logger.error("百度云人体检测与属性识别异常！error_code:{}, error_msg:{}, reply_num:{}, request_id: {}",
                                     enum.value[0], enum.value[2], reply_num, request_id)
                        raise Exception()
                    # 重试指定次数后，还是异常，输出对应的异常
                    if enum.value[3] == 1:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        raise ServiceException(str(enum.value[0]), enum.value[2])
                    # 重试指定次数后，还是异常，输出空
                    if enum.value[3] == 2:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        if reply_num >= reply_value:
                            return None
                        raise Exception()
                return res_image
            except Exception as e:
                time.sleep(0.5)
                reply_num += 1
                self.init_client()
                if reply_num > reply_value:
                    if isinstance(e, ServiceException):
                        raise ServiceException(e.code, e.msg)
                    logger.error("人体检测与属性识别失败: {}, request_id: {}", traceback.format_exc(), request_id)
                    raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                           ExceptionType.SERVICE_INNER_EXCEPTION.value[1])
        # except Exception as ee:
        #     logger.exception("车辆检测加锁异常: {}, request_id: {}", ee, request_id)
        #     raise ServiceException(ExceptionType.UNIVERSAL_TEXT_RECOGNITION_FAILED.value[0],
        #                            ExceptionType.UNIVERSAL_TEXT_RECOGNITION_FAILED.value[1])
        # finally:
        #     self.lock.release()

    '''
        人流量统计
    '''

    def bodyNum(self, url, request_id, options={}):
        self.init_client()
        image = self.readImage(url, request_id)
        # try:
        #     self.lock.acquire()
        reply_num = 0
        reply_value = None
        while True:
            try:
                options["show"] = "true"
                res_image = self.__client.bodyNum(image, options)
                error_code = res_image.get("error_code")
                if error_code:
                    enum = BAIDUERRORDATA.get(error_code)
                    # 如果异常编码未知, 返回空值
                    if enum is None:
                        logger.error("百度云人流量统计异常！error_code:{}, request_id: {}", error_code, request_id)
                        return None
                    # 重试指定次数后，还是异常，输出统一内部异常
                    if enum.value[3] == 0:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        logger.error("百度云人流量统计异常！error_code:{}, error_msg:{}, reply_num:{}, request_id: {}",
                                     enum.value[0], enum.value[2], reply_num, request_id)
                        raise Exception()
                    # 重试指定次数后，还是异常，输出对应的异常
                    if enum.value[3] == 1:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        raise ServiceException(str(enum.value[0]), enum.value[2])
                    # 重试指定次数后，还是异常，输出空
                    if enum.value[3] == 2:
                        if reply_value is None:
                            reply_value = enum.value[4]
                        if reply_num >= reply_value:
                            return None
                        raise Exception()
                return res_image
            except Exception as e:
                time.sleep(0.5)
                reply_num += 1
                self.init_client()
                if reply_num > reply_value:
                    if isinstance(e, ServiceException):
                        raise ServiceException(e.code, e.msg)
                    logger.exception("人流量统计失败: {}, request_id: {}", traceback.format_exc(), request_id)
                    raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                           ExceptionType.SERVICE_INNER_EXCEPTION.value[1])
        # except Exception as ee:
        #     logger.exception("车辆检测加锁异常: {}, request_id: {}", ee, request_id)
        #     raise ServiceException(ExceptionType.UNIVERSAL_TEXT_RECOGNITION_FAILED.value[0],
        #                            ExceptionType.UNIVERSAL_TEXT_RECOGNITION_FAILED.value[1])
        # finally:
        #     self.lock.release()

    def readImage(self, url, request_id):
        try:
            return url2Content(url)
        except Exception as e:
            logger.error("读取图片异常！url: {}, request_id: {}, 异常信息：{}", url, request_id, traceback.format_exc())
            raise ServiceException(ExceptionType.READ_IAMGE_URL_EXCEPTION.value[0],
                                   ExceptionType.READ_IAMGE_URL_EXCEPTION.value[1])

# if __name__ == '__main__':
#     with open(r"D:\work\alg_new\tuoheng_alg\dsp_dev_service.yml", "r", encoding='utf-8') as f:
#         file_content = f.read()
#     content = yaml.load(file_content, yaml.FullLoader)
#     aipImageClassifyClient = AipImageClassifyClient(content)
#     aipBodyAnalysisClient = AipBodyAnalysisClient(content)
#     url = "https://pic.52112.com/180623/JPG-180623-12/c4cyivkxEh_small.jpg"
#     # result = aipImageClassifyClient.vehicleDetectUrl(url, "1111111")
#     result = aipBodyAnalysisClient.bodyNum(url, "1111111")
#     iamge = base64.b64decode(result.get("image"))
#     print(iamge)