ASJServer/Controller/AlgorithmShop/AlgorithmShopController.py

# -*- encoding: utf-8 -*-
"""
@File    : AlgorithmShopController.py
@Time    : 2022/8/24 20:02
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import json
import logging
import math
import time
from datetime import datetime, timedelta

from django.db.models import F, Value, CharField, Sum
from django.views.generic.base import View

from Model.models import DeviceAlgorithmExplain, DeviceAlgorithmBanner, DeviceUidAlgorithmType, \
    DeviceTypeAlgorithmInfo, DeviceAppScenario, DeviceScenarioLangInfo, DeviceAlgorithmScenario, \
    DeviceAlgorithmPassengerFlow
from Object.ETkObject import ETkObject
from Object.Enums.TimeZone import TimeZone
from Object.RedisObject import RedisObject
from Object.ResponseObject import ResponseObject
from Object.TokenObject import TokenObject
from Object.utils import LocalDateTimeUtil

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):
        if operation == 'passengerFlowStatistical':
            return self.passenger_flow_statistical(request_dict, ResponseObject('en'))
        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)
        elif operation == 'getPassengerFlowList':  # 获取客流统计列表
            return self.get_passenger_flow_list(request_dict, response)
        else:
            return response.json(0)

    @classmethod
    def passenger_flow_statistical(cls, request_dict, response):
        """
        添加客流统计
        Args:
            request_dict (dict): 请求参数字典
            response: 响应对象
        """
        try:
            LOGGER.info('*****AlgorithmShopView.passenger_flow_statistical:params{}'.format(json.dumps(request_dict)))
            sign = request_dict.get('sign')
            eto = ETkObject(sign)
            uid = eto.uid
            if not uid:
                return response.json(444)

            d_time = request_dict.get('deviceTime')
            enter_count = request_dict.get('enterCount')
            tz_value = request_dict.get('timeZone')
            exit_count = request_dict.get('exitCount')

            if not all([sign, enter_count, exit_count, tz_value]):
                return response.json(444)
            enter_count = int(enter_count)
            exit_count = int(exit_count)
            redis = RedisObject(5)
            key = f'ASJ:PASSENGER:FLOW:{uid}:{d_time}'

            # 检查Redis中是否已存在相同key的数据
            r_data = redis.get_data(key)
            if r_data:
                return response.json(0)

            now_time = int(time.time())
            data = {
                'uid': uid,
                'updated_time': now_time,
                'created_time': now_time,
                'device_time': int(d_time)
            }

            tz = TimeZone.get_value(int(tz_value))
            date_time = LocalDateTimeUtil.time_format_date(int(d_time), tz)
            data['statistical_time'] = datetime.strptime(date_time, '%Y-%m-%d %H:%M:%S')

            LOGGER.info(f'uid{uid}-DeviceTime:{d_time},tz:{tz},结果：{date_time}')
            passenger_flow_list = []
            if enter_count > 0:
                data['count'] = enter_count
                data['type'] = 1
                passenger_flow_list.append(DeviceAlgorithmPassengerFlow(**data))
            if exit_count > 0:
                data['count'] = exit_count
                data['type'] = 2
                passenger_flow_list.append(DeviceAlgorithmPassengerFlow(**data))
            DeviceAlgorithmPassengerFlow.objects.bulk_create(passenger_flow_list)
            # 将数据存入Redis，并设置过期时间为10分钟
            redis.CONN.setnx(key, d_time)
            redis.CONN.expire(key, 600)

            return response.json(0)
        except Exception as e:
            LOGGER.info('***get_algorithm_list_by_scenario_id,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_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))

    @classmethod
    def get_passenger_flow_list(cls, request_dict, response):
        """
        获取客流统计列表
        """
        try:
            uid = request_dict.get('uid')
            now_time = request_dict.get('nowTime')
            flow_type = request_dict.get('flowType')
            query_type = request_dict.get('queryType')
            if not all([now_time, flow_type, query_type]):
                return response.json(444)
            query_type = int(query_type)
            flow_type = int(flow_type)
            passenger_flow_list = []
            # 将需要查询的时间戳列表拼接成缓存的键名
            cache_key = f"ASJ:PASSENGER_FLOW:{uid}_{flow_type}_{query_type}_{now_time}"
            redis = RedisObject(5)
            passenger_flow_json = redis.get_data(cache_key)
            if passenger_flow_json:
                return response.json(0, json.loads(passenger_flow_json))
            if query_type == 1:
                passenger_flow_list = cls.get_passenger_flow_by_day(uid, int(now_time), flow_type)
            elif query_type == 2:
                passenger_flow_list = cls.get_passenger_flow_by_month(uid, int(now_time), flow_type)
            elif query_type == 3:
                passenger_flow_list = cls.get_passenger_flow_by_year(uid, int(now_time), flow_type)
            else:
                return response.json(0, passenger_flow_list)
            if passenger_flow_list:
                redis.CONN.setnx(cache_key, json.dumps(passenger_flow_list))
                redis.CONN.expire(cache_key, 3600 * 24)
            return response.json(0, passenger_flow_list)
        except Exception as e:
            print('查询客流异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
            return response.json(500, repr(e))

    @staticmethod
    def get_passenger_flow_by_day(uid, now_time, flow_type):
        """
        按天获取客流统计
        """
        c_time = datetime.fromtimestamp(int(now_time))
        # 一天周期24小时从当前小时整点往前查询近24小时，生成秒级时间戳
        t_list = [math.floor((c_time - timedelta(hours=i)).timestamp()) for i in range(24)]
        passenger_flow_list = []
        pf_qs = DeviceAlgorithmPassengerFlow.objects.filter(
            device_time__in=t_list,
            type=int(flow_type),
            uid=uid
        ).order_by('device_time')
        if not pf_qs.exists():
            return passenger_flow_list
        # 将查询到的数据时间戳转字典
        pf_dict = {pf.device_time: pf for pf in pf_qs}
        # 循环查找最近24小时的记录（包含当前时间）
        for i in range(24):
            current_timestamp = t_list[i]
            if current_timestamp in pf_dict:
                item = pf_dict[current_timestamp]
                passenger_flow_list.append({
                    'index': 23 - i,
                    'count': item.count,
                    'time': item.device_time
                })
        return passenger_flow_list

    @staticmethod
    def get_passenger_flow_by_month(uid, now_time, flow_type):
        """
        按月获取客流统计
        """

        # 获取30天前的时间戳作为起始时间
        s_time = LocalDateTimeUtil.get_before_days_timestamp(now_time, 30)
        # 将起始时间设置为当天的开始时间
        s_time = LocalDateTimeUtil.start_of_day_timestamp(s_time)

        # 查询指定类型的客流数据，按设备时间升序排序
        pf_qs = DeviceAlgorithmPassengerFlow.objects.filter(
            device_time__range=(s_time, now_time),
            type=int(flow_type),
            uid=uid
        ).order_by('device_time')

        passenger_flow_list = []
        if not pf_qs.exists():
            return passenger_flow_list

        for i in range(30):  # 查询月按30天为一个周期
            if i == 0:
                # 当天的开始时间作为查询的起始时间
                s_time = LocalDateTimeUtil.start_of_day_timestamp(now_time)
            else:
                # 获取i天前的时间戳作为查询的起始时间
                s_time = LocalDateTimeUtil.get_before_days_timestamp(now_time, i)
                # 将起始时间设置为当天的开始时间
                s_time = LocalDateTimeUtil.start_of_day_timestamp(s_time)

            days_qs = []

            for item in pf_qs:
                # 计算结束时间
                end_time = now_time if i == 0 else s_time + (3600 * 24)

                # 将在起始时间和结束时间范围内的客流数据放入列表中
                if s_time < item.device_time <= end_time:
                    days_qs.append(item.count)

            if days_qs:
                # 将每天的客流数据统计结果添加到列表中
                passenger_flow_list.append({
                    'index': 29 - i,
                    'count': sum(days_qs),
                    'time': s_time
                })

        return passenger_flow_list

    @staticmethod
    def get_passenger_flow_by_year(uid, now_time, flow_type):
        """
        按年获取客流统计
        """
        month_time = 0
        passenger_flow_list = []
        for i in range(12):
            if i == 0:
                month_time = LocalDateTimeUtil.get_current_month_first_day(now_time)
            df_qs = DeviceAlgorithmPassengerFlow.objects.filter(type=flow_type, uid=uid)
            if i == 0:
                df_qs = df_qs.filter(device_time__range=(month_time, now_time))
            else:
                previous_month_time = LocalDateTimeUtil.get_previous_month_first_day(month_time)
                df_qs = df_qs.filter(device_time__range=(previous_month_time, month_time))
            df_qs = df_qs.aggregate(total=Sum('count'))
            if not df_qs['total']:
                continue

            passenger_flow_list.append({
                'index': 11 - i,
                'count': df_qs['total'],
                'time': month_time
            })
        return passenger_flow_list