# -*- encoding: utf-8 -*- """ @File : AlgorithmShopController.py @Time : 2022/8/24 20:02 @Author : stephen @Email : zhangdongming@asj6.wecom.work @Software: PyCharm """ import logging import time from django.db.models import F, Value, CharField from django.views.generic.base import View from Model.models import DeviceAlgorithmExplain, DeviceAlgorithmBanner, DeviceUidAlgorithmType, \ DeviceTypeAlgorithmInfo, DeviceAppScenario, DeviceScenarioLangInfo, DeviceAlgorithmScenario from Object.ResponseObject import ResponseObject from Object.TokenObject import TokenObject LOGGER = logging.getLogger('info') class AlgorithmShopView(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.GET, request, operation) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.POST, request, operation) def validation(self, request_dict, request, operation): token = TokenObject(request.META.get('HTTP_AUTHORIZATION')) lang = request_dict.get('lang', token.lang) response = ResponseObject(lang) if token.code != 0: return response.json(token.code) if operation == 'list': return self.algorithm_list(request_dict, response) elif operation == 'banner-list': return self.get_algorithm_banner(response) elif operation == 'uid-details': return self.get_algorithm_details(request_dict, response) elif operation == 'save': return self.algorithm_setting_save(request_dict, response) elif operation == 'getScenarioList': # 获取应用场景数据列表 return self.get_scenario_list(request_dict, response) elif operation == 'getAlgorithmListByScenarioId': # 根据应用场景id获取算法列表 return self.get_scenario_algorithm_list(request_dict, response) else: return response.json(0) @classmethod def get_algorithm_list_by_scenario_id(cls, scenario_id, lang): """ 根据应用场景ID查询算法信息列表 @param scenario_id: 场景ID @param lang: 语言 @return: 算法类型信息 """ try: if not scenario_id or scenario_id == 0: return [] # 根据场景id查询关联的算法id algorithm_scenario_qs = DeviceAlgorithmScenario.objects.filter(scenario_id=scenario_id) \ .order_by('sort').values('algorithm_id') if not algorithm_scenario_qs.exists(): return [] algorithm_list = [] for item in algorithm_scenario_qs: algorithm_id = item['algorithm_id'] # 根据算法id查询多语言数据 algorithm_list.append(cls.get_lang_info_by_algorithm_id(algorithm_id, lang, None)) return algorithm_list except Exception as e: LOGGER.info('***get_algorithm_list_by_scenario_id,errLine:{}, errMsg:{}' .format(e.__traceback__.tb_lineno, repr(e))) return [] @classmethod def get_lang_info_by_algorithm_id(cls, algorithm_id, lang, uid): """ 根据算法id查询多语言数据详情 @param uid: 设备uid @param algorithm_id: 算法id @param lang: 语言 @return: 算法多语言数据详情 """ try: algorithm_qs = DeviceAlgorithmExplain.objects.filter(algorithm_type_id=algorithm_id, lang=lang) \ .values('algorithm_type__icon_url', 'algorithm_type__id', 'title', 'subtitle', 'algorithm_type__image_url', 'algorithm_type__basic_function', 'concerning', 'price', 'algorithm_type__tag', 'algorithm_type__status', 'algorithm_type__type') if not algorithm_qs.exists(): return {} setting = '' # 当前支持设置的算法功能json # 存在uid则查询当前uid是否支持该算法 if uid: setting = cls.get_uid_algorithm_info(algorithm_id, uid) setting = setting if setting else {'status': 0, 'function': {}} data = { 'typeId': algorithm_qs[0]['algorithm_type__id'], 'iconUrl': algorithm_qs[0]['algorithm_type__icon_url'], 'imageUrl': algorithm_qs[0]['algorithm_type__image_url'], 'title': algorithm_qs[0]['title'], 'subtitle': algorithm_qs[0]['subtitle'], 'basicFunction': algorithm_qs[0]['algorithm_type__basic_function'], 'concerning': algorithm_qs[0]['concerning'], 'price': algorithm_qs[0]['price'], 'tag': algorithm_qs[0]['algorithm_type__tag'], 'status': algorithm_qs[0]['algorithm_type__status'], 'setting': setting, 'type': algorithm_qs[0]['algorithm_type__type'] } return data except Exception as e: LOGGER.info('***get_lang_info_by_algorithm_id,errLine:{}, errMsg:{}' .format(e.__traceback__.tb_lineno, repr(e))) return {} @classmethod def get_algorithm_list(cls, lang): """ 获取所有算法数据列表 @return: 算法数据列表 """ algorithm_qs = DeviceAlgorithmExplain.objects.filter(lang=lang).order_by('algorithm_type__sort') \ .annotate(iconUrl=F('algorithm_type__icon_url'), typeId=F('algorithm_type__id'), type=F('algorithm_type__type'), imageUrl=F('algorithm_type__image_url'), basicFunction=F('algorithm_type__basic_function'), tag=F('algorithm_type__tag'), status=F('algorithm_type__status'), setting=Value('', output_field=CharField())) \ .values('iconUrl', 'imageUrl', 'title', 'subtitle', 'concerning', 'basicFunction', 'price', 'tag', 'status', 'setting', 'typeId', 'type') if not algorithm_qs.exists(): return [] return list(algorithm_qs) @classmethod def get_scenario_list(cls, request_dist, response): """ 获取应用场景列表 @param request_dist: lang @param response: 响应结果 @return: 应用场景列表 """ try: lang = request_dist.get('lang', 'en') if not lang: return response.json(444) # 获取应用场景列表 scenario_qs = DeviceAppScenario.objects.filter().exclude(type=0).all().order_by('sort') \ .values('id', 'type', 'cver_url', 'banner_url') scenario_list = [] if not scenario_qs.exists(): return response.json(0, scenario_list) for item in scenario_qs: scenario_vo = {'id': item['id'], 'cverUrl': item['cver_url'], 'bannerUrl': item['banner_url'], 'name': '', 'content': ''} # 获取根据语言应用场景信息 scenario_info_qs = DeviceScenarioLangInfo.objects.filter(lang=lang, scenario_id=item['id']) \ .values('name', 'content') if not scenario_info_qs.exists(): continue scenario_vo['name'] = scenario_info_qs[0]['name'] scenario_vo['content'] = scenario_info_qs[0]['content'] # 根据应用场景id查询关联算法类型数据 # scenario_vo['algorithmList'] = cls.get_algorithm_list_by_scenario_id(item['id'], lang) scenario_list.append(scenario_vo) # 获取所有算法图标地址以及算法名称 algorithm_qs = DeviceAlgorithmExplain.objects.filter(lang=lang).order_by('algorithm_type__sort') \ .annotate(algorithmId=F('algorithm_type__id'), algorithmType=F('algorithm_type__type'), iconUrl=F('algorithm_type__icon_url'), algorithmName=F('title')).values('algorithmId', 'algorithmType', 'iconUrl', 'algorithmName') scenario_qs = DeviceAppScenario.objects.filter(type=0) \ .values('cver_url', 'banner_url') scenario_banner = {} if scenario_qs.exists(): scenario_banner['cverUrl'] = scenario_qs[0]['cver_url'] scenario_banner['bannerUrl'] = scenario_qs[0]['banner_url'] result_dto = {'scenarioList': scenario_list, 'scenarioUrl': scenario_banner} if algorithm_qs.exists(): result_dto['iconList'] = list(algorithm_qs) return response.json(0, result_dto) except Exception as e: LOGGER.info('接口异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @classmethod def get_scenario_algorithm_list(cls, request_dist, response): """ 获取应用场景关联算法列表 @param request_dist: scenarioId、lang @param response: 响应结果 @return: 算法列表 """ scenario_id = request_dist.get('scenarioId', None) lang = request_dist.get('lang', 'en') result_dto = {'scenarioBannerUrl': ''} if not scenario_id: result_dto['algorithmList'] = cls.get_algorithm_list(lang) return response.json(0, result_dto) result_dto['algorithmList'] = cls.get_algorithm_list_by_scenario_id(scenario_id, lang) return response.json(0, result_dto) @classmethod def get_algorithm_banner(cls, response): """ 获取算法小店banner """ banner_qs = DeviceAlgorithmBanner.objects.all() banner_vs = banner_qs.order_by('sort') \ .values('algorithm_type__type', 'algorithm_type__id', 'image_url') banner_list = [] if not banner_vs.exists(): return response.json(0, banner_list) for item in banner_vs: banner_list.append({ 'typeId': item['algorithm_type__id'], 'type': item['algorithm_type__type'], 'imageUrl': item['image_url'], }) return response.json(0, banner_list) @classmethod def algorithm_list(cls, request_dict, response): """ 获取算法小店列表 """ try: lang = request_dict.get('lang', 'en') uid = request_dict.get('uid', None) version = request_dict.get('version', 'v1') algorithm_qs = DeviceAlgorithmExplain.objects.filter(lang=lang) types = [0, 1, 3, 4, 5] if version == 'v1': algorithm_qs = algorithm_qs.filter(algorithm_type__type__in=types) algorithm_qs = algorithm_qs.order_by('algorithm_type__sort') \ .values('algorithm_type__id', 'algorithm_type__type', 'algorithm_type__icon_url', 'title', 'subtitle', 'algorithm_type__image_url', 'algorithm_type__basic_function', 'concerning') algorithm_list = [] if not algorithm_qs.exists(): return response.json(0, algorithm_list) for item in algorithm_qs: setting = '' if uid: setting = cls.get_uid_algorithm_info(item['algorithm_type__id'], uid) setting = setting if setting else {'status': 0, 'function': {}} algorithm_list.append({ 'typeId': item['algorithm_type__id'], 'type': item['algorithm_type__type'], 'iconUrl': item['algorithm_type__icon_url'], 'imageUrl': item['algorithm_type__image_url'], 'title': item['title'], 'subtitle': item['subtitle'], 'setting': setting, 'basicFunction': item['algorithm_type__basic_function'], 'concerning': item['concerning'] }) return response.json(0, algorithm_list) except Exception as e: print('查询算法小店列表异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @classmethod def get_algorithm_details(cls, request_dict, response): """ 获取算法小店类型详情 """ try: lang = request_dict.get('lang', 'en') type_id = request_dict.get('typeId', None) if not type_id: return response.json(444, 'typeId not null') type_id = int(type_id) uid = request_dict.get('uid', None) explain_qs = DeviceAlgorithmExplain.objects.filter(lang=lang).filter(algorithm_type__id=type_id) \ .values('algorithm_type__id', 'algorithm_type__type', 'algorithm_type__down_count', 'algorithm_type__details_img_url', 'algorithm_type__icon_url', 'title', 'subtitle', 'introduction', 'install_explain', 'risk_warning', 'algorithm_type__basic_function', 'concerning') if not explain_qs.exists(): return response.json(0, {}) item = explain_qs.first() algorithm_dict = { 'typeId': item['algorithm_type__id'], 'type': item['algorithm_type__type'], 'downCount': item['algorithm_type__down_count'], 'detailsImgUrl': item['algorithm_type__details_img_url'], 'iconUrl': item['algorithm_type__icon_url'], 'title': item['title'], 'subtitle': item['subtitle'], 'introduction': item['introduction'], 'installExplain': item['install_explain'], 'riskWarning': item['risk_warning'], 'basicFunction': item['algorithm_type__basic_function'], 'concerning': item['concerning'] } dt_info_qs = DeviceTypeAlgorithmInfo.objects.filter(algorithm_type=algorithm_dict['type']) \ .annotate(deviceName=F('device_name'), deviceType=F('device_type'), algorithmType=F('algorithm_type'), typeIcon=F('type_icon'), deviceLink=F('device_link'), ) \ .values('deviceName', 'deviceType', 'typeIcon', 'deviceLink') algorithm_dict['recommendDevices'] = list(dt_info_qs) if uid: setting = cls.get_uid_algorithm_info(item['algorithm_type__id'], uid) algorithm_dict['setting'] = setting if setting else {'status': 0, 'function': {}} return response.json(0, algorithm_dict) except Exception as e: print('查询算法详情异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return response.json(177, repr(e)) @staticmethod def get_uid_algorithm_info(type_id, uid): """ 获取当前设备使用算法状态信息 @param type_id: 算法类型ID @param uid: 设备唯一标识 @return: dict """ uid_algorithm_qs = DeviceUidAlgorithmType.objects.filter(algorithm_type_id=type_id, device_uid=uid) \ .values('status', 'function') if not uid_algorithm_qs.exists(): return None return uid_algorithm_qs.first() @classmethod def algorithm_setting_save(cls, request_dict, response): """ 算法设置保存 """ try: type_id = request_dict.get('typeId', None) uid = request_dict.get('uid', None) status = request_dict.get('status', None) setting_json = request_dict.get('function') if not all([type_id, uid, status, setting_json]): return response.json(444) status = int(status) type_id = int(type_id) now_time = int(time.time()) uid_algorithm_qs = DeviceUidAlgorithmType.objects.filter(algorithm_type_id=type_id, device_uid=uid) if not uid_algorithm_qs.exists(): param = {'algorithm_type_id': int(type_id), 'uid': uid, 'function': setting_json, 'status': status, 'updated_time': now_time, 'created_time': now_time} DeviceUidAlgorithmType.objects.create(**param) return response.json(0) uid_algorithm_qs.update(status=status, function=setting_json, updated_time=now_time) return response.json(0) except Exception as e: print('保存算法设置异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return response.json(177, repr(e))