# -*- 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)