ASJPush/Controller/gatewayController.py

# -*- coding: utf-8 -*-
"""
@Author ： Rocky
@Time ： 2022/5/9 10:51
@File ：gatewayController.py
"""
import datetime
import threading
import time

import pytz
import requests
from django.views.generic.base import View

from AnsjerPush.Config.gatewaySensorConfig import SENSOR_TYPE, EVENT_TYPE, DEVICE_TYPE, SMART_SOCKET_TOPIC
from AnsjerPush.config import LOGGER, XM_PUSH_CHANNEL_ID
from Model.models import SensorRecord, GatewaySubDevice, GatewayPush, Device_Info, SceneLog, SmartScene, CountryModel, \
    EffectiveTime
from Object.ResponseObject import ResponseObject
from Service.CommonService import CommonService
from AnsjerPush.config import DETECT_PUSH_DOMAIN
from Service.EquipmentInfoService import EquipmentInfoService
from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject
from Service.PushService import PushObject


class GatewayView(View):
    def get(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.GET, operation)

    def post(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.POST, operation)

    def validation(self, request_dict, operation):
        response = ResponseObject()
        if operation == 'gatewayPush':  # 网关推送
            return self.gateway_push(request_dict, response)
        elif operation == 'sceneLogPush':  # 场景日志推送
            return self.scene_log_push(request_dict, response)
        elif operation == 'socketPush':  # 插座推送
            return self.socket_msg_push(request_dict, response)
        else:
            return response.json(414)

    @classmethod
    def gateway_push(cls, request_dict, response):
        """
        网关推送
        @param request_dict: 请求参数
        @request_dict serial_number: 序列号
        @request_dict ieee_addr: 长地址
        @request_dict sensor_type: 传感器类型
        @request_dict event_type: 事件类型
        @request_dict defense: 防御状态,0:撤防,1:防御
        @request_dict sensor_status: 拆动状态,拆动时传参
        @param response: 响应对象
        @return: response
        """
        serial_number = request_dict.get('serial_number', None)
        ieee_addr = request_dict.get('ieee_addr', None)
        sensor_type = int(request_dict.get('sensor_type', None))
        event_type = int(request_dict.get('event_type', None))
        defense = int(request_dict.get('defense', None))
        LOGGER.info('---调用网关推送接口--- request_dict:{}'.format(request_dict))
        if not all([serial_number, ieee_addr, sensor_type, event_type]):
            return response.json(444)

        num = None
        n_time = int(time.time())

        try:
            # 获取温湿度
            if sensor_type == SENSOR_TYPE['tem_hum_sensor'] and (
                    event_type == EVENT_TYPE['temperature'] or event_type == EVENT_TYPE['humidity']):
                num = request_dict.get('num')
                if num is None:
                    return response.json(444)
                num = int(num) / 100

            # 查询子设备表id
            gateway_sub_device_qs = GatewaySubDevice.objects.filter(device__serial_number=serial_number,
                                                                    device_type=sensor_type, ieee_addr=ieee_addr). \
                values('id', 'nickname', 'device__userID__region_country')
            if not gateway_sub_device_qs.exists():
                return response.json(173)

            gateway_sub_device_id = gateway_sub_device_qs[0]['id']
            # 多线程执行场景任务
            task_kwargs = {
                'num': num,
                'n_time': n_time,
                'event_type': event_type,
                'gateway_sub_device_id': gateway_sub_device_id
            }
            execute_task_thread = threading.Thread(target=cls.execute_scene_tasks, kwargs=task_kwargs)
            execute_task_thread.start()

            country_id = gateway_sub_device_qs[0]['device__userID__region_country']
            lang = cls.confirm_lang(country_id)
            alarm = cls.get_alarm(lang, event_type)

            nickname = gateway_sub_device_qs[0]['nickname']
            sensor_record_dict = {
                'gateway_sub_device_id': gateway_sub_device_id,
                'alarm': alarm,
                'event_type': event_type,
                'created_time': n_time,
            }
            # 温湿度上报不推送
            if num is not None:
                sensor_record_dict['alarm'] = str(num)
                SensorRecord.objects.create(**sensor_record_dict)
                return response.json(0)

            SensorRecord.objects.create(**sensor_record_dict)

            # 门磁被拆动/拆动恢复,修改拆动状态
            if event_type == 2156:
                gateway_sub_device_qs.update(is_tampered=1)
            elif event_type == 2152:
                gateway_sub_device_qs.update(is_tampered=0)

            # 撤防状态不推送
            if defense == 0:
                return response.json(0)

            device_info_qs = Device_Info.objects.filter(serial_number=serial_number).values('userID_id')
            if not device_info_qs.exists():
                return response.json(173)

            equipment_info_list = []
            equipment_info_model = EquipmentInfoService.randoms_choice_equipment_info()
            # 推送表存储数据
            equipment_info_kwargs = {
                'device_uid': serial_number,
                'device_nick_name': nickname,
                'event_type': event_type,
                'event_time': n_time,
                'add_time': n_time,
                'alarm': alarm
            }
            for device_info in device_info_qs:
                user_id = device_info['userID_id']
                equipment_info_kwargs['device_user_id'] = user_id
                equipment_info_list.append(equipment_info_model(**equipment_info_kwargs))

                # 开启异步推送
                push_kwargs = {
                    'user_id': user_id,
                    'n_time': n_time,
                    'event_type': event_type,
                    'nickname': nickname,
                    'alarm': alarm,
                }
                push_thread = threading.Thread(
                    target=cls.gateway_push_msg,
                    kwargs=push_kwargs)
                push_thread.start()

            if equipment_info_list:
                equipment_info_model.objects.bulk_create(equipment_info_list)

            return response.json(0)
        except Exception as e:
            LOGGER.info('---网关推送接口异常--- {}'.format(repr(e)))
            return response.json(500, 'error_ine:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    @staticmethod
    def confirm_lang(country_id):
        """
        根据country_id确定语言
        @param country_id: 国家id
        @return lang: 语言
        """
        country_qs = CountryModel.objects.filter(id=country_id).values('country_code')
        if not country_qs.exists():
            lang = 'NA'
        else:
            lang = country_qs[0]['country_code']
        return lang

    @staticmethod
    def get_alarm(lang, event_type):
        """
        根据语言和事件类型确定警报内容
        @param lang: 语言
        @param event_type: 事件类型
        @return alarm: 警报内容
        """
        alarm = ''
        if lang == 'CN':
            # 门磁
            if event_type == 2150:
                alarm = '门磁开'
            elif event_type == 2151:
                alarm = '门磁关'
            elif event_type == 2156:
                alarm = '被拆动'
            elif event_type == 2152:
                alarm = '拆动恢复'

            # 智能按钮
            elif event_type == 2160:
                alarm = '紧急按钮按下'
            elif event_type == 2161:
                alarm = '单击'
            elif event_type == 2162:
                alarm = '双击'
            elif event_type == 2163:
                alarm = '长按'

            # 水浸
            elif event_type == 2170:
                alarm = '水浸触发'
            elif event_type == 2171:
                alarm = '水浸恢复'

            # 烟雾
            elif event_type == 2180:
                alarm = '烟雾触发'
            elif event_type == 2181:
                alarm = '烟雾恢复'

            # 人体红外
            elif event_type == 2190:
                alarm = '有人移动'
            elif event_type == 2191:
                alarm = '无人移动'

            # 低电量
            elif event_type in (2153, 2164, 2172, 2182, 2193):
                alarm = '低电量'
        else:
            # 门磁
            if event_type == 2150:
                alarm = 'Door magnetic opening'
            elif event_type == 2151:
                alarm = 'Door magnetic closing'
            elif event_type == 2156:
                alarm = 'Be dismantled'
            elif event_type == 2152:
                alarm = 'Dismantling recovery'

            # 智能按钮
            elif event_type == 2160:
                alarm = 'Emergency button pressed'
            elif event_type == 2161:
                alarm = 'Single click'
            elif event_type == 2162:
                alarm = 'Double click'
            elif event_type == 2163:
                alarm = 'Long press'

            # 水浸
            elif event_type == 2170:
                alarm = 'Water immersion trigger'
            elif event_type == 2171:
                alarm = 'Water immersion recovery'

            # 烟雾
            elif event_type == 2180:
                alarm = 'Smoke triggering'
            elif event_type == 2181:
                alarm = 'Smoke recovery'

            # 人体红外
            elif event_type == 2190:
                alarm = 'Someone moving'
            elif event_type == 2191:
                alarm = 'Unmanned movement'

            # 低电量
            elif event_type in (2153, 2164, 2172, 2182, 2193):
                alarm = 'LOW BATTERY'
        return alarm

    @classmethod
    def scene_log_push(cls, request_dict, response):
        """
        网关智能场景日志推送
        @param request_dict: 请求参数
        @request_dict sceneId: 场景id
        @request_dict status: 状态
        @param response: 响应对象
        @return: response
        """
        scene_id = request_dict.get('sceneId', None)
        status = request_dict.get('status', None)

        LOGGER.info('---场景日志推送接口--- request_dict:{}'.format(request_dict))
        if not all([scene_id, status]):
            return response.json(444)

        smart_scene_qs = SmartScene.objects.filter(id=scene_id).values('scene_name', 'conditions', 'tasks', 'device_id',
                                                                       'sub_device_id', 'user_id', 'scene_data')
        if not smart_scene_qs.exists():
            return response.json(173)

        scene_name = smart_scene_qs[0]['scene_name']
        tasks = smart_scene_qs[0]['tasks']
        device_id = smart_scene_qs[0]['device_id']
        sub_device_id = smart_scene_qs[0]['sub_device_id']
        n_time = int(time.time())
        user_id = smart_scene_qs[0]['user_id']
        scene_data = smart_scene_qs[0]['scene_data']
        if sub_device_id:
            gateway_sub_device_qs = GatewaySubDevice.objects.filter(id=sub_device_id).values('nickname')
            nickname = gateway_sub_device_qs[0]['nickname'] if gateway_sub_device_qs.exists() else ''
        else:
            device_qs = Device_Info.objects.filter(id=device_id).values('NickName')
            nickname = device_qs[0]['NickName'] if device_qs.exists() else ''
        log_dict = {
            'scene_id': scene_id,
            'scene_name': scene_name,
            'tasks': tasks,
            'status': status,
            'device_id': device_id,
            'sub_device_id': sub_device_id,
            'created_time': n_time,
        }
        tasks = eval(tasks)

        try:
            SceneLog.objects.create(**log_dict)
            # 如果是一次性场景,关闭场景
            if scene_data:
                scene_data_dict = eval(scene_data)
                condition = scene_data_dict.get('condition')
                if condition:
                    time_type = condition.get('time')
                    if time_type == 'date':
                        smart_scene_qs.update(is_enable=False)

            # 推送日志
            gateway_push_qs = GatewayPush.objects.filter(user_id=user_id, logout=False). \
                values('user_id', 'app_bundle_id', 'app_type', 'push_type', 'token_val', 'm_code', 'lang', 'm_code',
                       'tz')
            if not gateway_push_qs.exists():
                return response.json(174)

            # 开启异步推送
            push_kwargs = {
                'tasks': tasks,
                'n_time': n_time,
                'nickname': nickname,
                'gateway_push_qs': gateway_push_qs
            }
            push_thread = threading.Thread(
                target=cls.scene_msg_push,
                kwargs=push_kwargs)
            push_thread.start()

            return response.json(0)
        except Exception as e:
            LOGGER.info('---场景日志推送接口异常--- {}'.format(repr(e)))
            return response.json(500, 'error_ine:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    @classmethod
    def gateway_push_msg(cls, **push_kwargs):
        user_id = push_kwargs['user_id']
        # 查询推送配置数据
        gateway_push_qs = GatewayPush.objects.filter(user_id=user_id, logout=False). \
            values('user_id', 'app_bundle_id', 'app_type', 'push_type', 'token_val', 'm_code', 'lang', 'm_code',
                   'tz')
        if gateway_push_qs.exists():
            n_time = push_kwargs['n_time']
            event_type = push_kwargs['event_type']
            nickname = push_kwargs['nickname']
            alarm = push_kwargs['alarm']

            kwargs = {
                'n_time': n_time,
                'event_type': event_type,
                'nickname': nickname,
            }

            # 推送到每台登录账号的手机
            for gateway_push in gateway_push_qs:
                app_bundle_id = gateway_push['app_bundle_id']
                push_type = gateway_push['push_type']
                token_val = gateway_push['token_val']
                lang = gateway_push['lang']
                tz = gateway_push['tz'] if gateway_push['tz'] else 0

                # 获取推送所需数据
                msg_title = PushObject.get_msg_title(nickname)
                msg_text = PushObject.get_gateway_msg_text(n_time, tz, lang, alarm)

                kwargs['msg_title'] = msg_title
                kwargs['msg_text'] = msg_text
                kwargs['app_bundle_id'] = app_bundle_id
                kwargs['token_val'] = token_val

                try:
                    # 推送消息
                    cls.push_msg(push_type, **kwargs)
                except Exception as e:
                    LOGGER.info('网关推送消息异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
                    continue

    @classmethod
    def scene_msg_push(cls, **push_kwargs):
        tasks = push_kwargs['tasks']
        n_time = push_kwargs['n_time']
        nickname = push_kwargs['nickname']
        gateway_push_qs = push_kwargs['gateway_push_qs']
        for task in tasks:
            event_type = task['event_type']
            if event_type == '1001':
                kwargs = {
                    'n_time': n_time,
                    'event_type': event_type,
                    'nickname': nickname,
                }
                event_info = task['value']
                # 推送到每台登录账号的手机
                for gateway_push in gateway_push_qs:
                    app_bundle_id = gateway_push['app_bundle_id']
                    push_type = gateway_push['push_type']
                    token_val = gateway_push['token_val']

                    kwargs['msg_title'] = PushObject.get_msg_title(nickname)
                    kwargs['msg_text'] = event_info
                    kwargs['app_bundle_id'] = app_bundle_id
                    kwargs['token_val'] = token_val

                    try:
                        # 推送消息
                        cls.push_msg(push_type, **kwargs)
                    except Exception as e:
                        LOGGER.info(
                            '场景日志推送消息异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
                        continue

    @staticmethod
    def push_msg(push_type, **kwargs):
        """
        发送推送消息
        @param push_type: 推送类型
        @param kwargs: 推送参数
        @return: None
        """
        if push_type == 0:  # ios apns
            PushObject.ios_apns_push(**kwargs)
        elif push_type == 1:  # android gcm
            PushObject.android_fcm_push(**kwargs)
        elif push_type == 2:  # android 极光推送
            PushObject.android_jpush(**kwargs)
        elif push_type == 3:
            huawei_push_object = HuaweiPushObject()
            huawei_push_object.send_push_notify_message(**kwargs)
        elif push_type == 4:  # android 小米推送
            channel_id = XM_PUSH_CHANNEL_ID['device_reminder']
            PushObject.android_xmpush(channel_id=channel_id, **kwargs)
        elif push_type == 5:  # android vivo推送
            PushObject.android_vivopush(**kwargs)
        elif push_type == 6:  # android oppo推送
            channel_id = 'DEVICE_REMINDER'
            PushObject.android_oppopush(channel_id=channel_id, **kwargs)
        elif push_type == 7:  # android 魅族推送
            PushObject.android_meizupush(**kwargs)

    @classmethod
    def socket_msg_push(cls, request_dict, response):
        """
        智能插座开关状态推送
        """
        try:
            serial_number = request_dict.get('serialNumber', None)
            device_time = request_dict.get('deviceTime', None)
            status = request_dict.get('status', None)
            if not all([serial_number, status, device_time]):
                return response.json(444)
            status = int(status)
            now_time = int(device_time) if device_time else int(time.time())
            # 获取主用户设备id
            log_dict = {
                'status': status,
                'device_id': serial_number,
                'created_time': now_time,
            }
            SceneLog.objects.create(**log_dict)
            LOGGER.info('成功接收并保存,插座序列号{},状态：{}'.format(serial_number, status))
            return response.json(0)
        except Exception as e:
            LOGGER.info('插座开关日志推送接口异常, error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    @classmethod
    def execute_scene_tasks(cls, gateway_sub_device_id, event_type, num, n_time):
        """
        执行场景任务
        @param gateway_sub_device_id: 子设备id
        @param event_type: 事件类型
        @param num: 温湿度值
        @param n_time: 当前时间
        @return:
        """
        smart_scene_qs = SmartScene.objects.filter(sub_device_id=gateway_sub_device_id, is_enable=True). \
            values('id', 'scene_data', 'tz', 'is_all_day', 'effective_time_id')
        if smart_scene_qs.exists():
            for smart_scene in smart_scene_qs:
                if smart_scene['scene_data']:
                    scene_data_dict = eval(smart_scene['scene_data'])
                    condition = scene_data_dict.get('condition')
                    if condition:
                        # None 或 event_type
                        scene_data_event_type = condition.get('event_type')
                        # 触发事件类型跟条件事件类型一致,且任务列表不为空
                        task_list = scene_data_dict.get('task_list')
                        if event_type == scene_data_event_type and task_list:
                            # 判断温湿度是否在条件设置的范围
                            if num is not None:
                                condition_value = condition.get('value')
                                space_index = condition_value.index(' ')
                                symbol, value = condition_value[:space_index], float(condition_value[space_index + 1:])
                                # 温湿度不在设定范围,不执行
                                if not ((symbol == '≥' and num >= value) or (symbol == '≤' and num <= value)):
                                    continue
                            execute_task = cls.judge_execute_task(smart_scene, n_time)

                            # 执行任务
                            if execute_task:
                                scene_id = 0
                                task_list_len = len(task_list)
                                for index, task in enumerate(task_list):
                                    # 最后一个任务上报日志
                                    if index == task_list_len - 1:
                                        scene_id = smart_scene['id']
                                    kwargs = {
                                        'device_type': task['device_type'],
                                        'event_type': task['event_type'],
                                        'serial_number': task['serial_number'],
                                        'delay_time': task['delay_time'],
                                        'scene_id': scene_id
                                    }
                                    pub_mqtt_thread = threading.Thread(target=cls.pub_mqtt, kwargs=kwargs)
                                    pub_mqtt_thread.start()

    @classmethod
    def judge_execute_task(cls, smart_scene, n_time):
        """
        判断是否执行任务
        @param smart_scene: 场景数据
        @param n_time: 当前时间
        @return execute_task: bool
        """
        execute_task = False
        # 判断时间是否在执行时间范围内
        if smart_scene['is_all_day'] == 1:
            # 全天必执行
            execute_task = True
        elif smart_scene['is_all_day'] == 2:
            # 非全天,判断当前时间是否在设置的时间范围
            tz = smart_scene['tz']
            time_minute, week = cls.get_time_info(n_time, tz)

            effective_time_id = smart_scene['effective_time_id']
            effective_time_qs = EffectiveTime.objects.filter(id=effective_time_id). \
                values('start_time', 'end_time', 'repeat')
            if effective_time_qs.exists():
                start_time = effective_time_qs[0]['start_time']
                end_time = effective_time_qs[0]['end_time']
                repeat = effective_time_qs[0]['repeat']
                time_frame_dict, time_frame_next_day_dict = cls.get_time_frame_dict(
                    start_time, end_time, repeat)
                # 判断当前时间是否在设置的时间范围
                if time_frame_dict.get(week):
                    start_time = time_frame_dict[week]['start_time']
                    end_time = time_frame_dict[week]['end_time']
                    if start_time <= time_minute <= end_time:
                        execute_task = True
                if time_frame_next_day_dict.get(week):
                    start_time = time_frame_next_day_dict[week]['start_time']
                    end_time = time_frame_next_day_dict[week]['end_time']
                    if start_time <= time_minute <= end_time:
                        execute_task = True
        return execute_task

    @staticmethod
    def get_time_info(timestamp, timezone_offset):
        """
        根据时间戳和时区获取时间(时分转为分钟数)和星期
        @param timestamp: 时间戳
        @param timezone_offset: 时区
        @return: time_minute, week
        """
        # 计算时区偏移量（以分钟为单位）
        timezone_offset_minutes = int(timezone_offset * 60)
        timezone = pytz.FixedOffset(timezone_offset_minutes)
        # 将时间戳转换为datetime对象，并应用时区
        dt_object = datetime.datetime.fromtimestamp(timestamp, tz=timezone)
        # 获取时分,并转为分钟数
        hour = dt_object.hour
        minute = dt_object.minute
        time_minute = hour * 60 + minute
        # 获取星期（0表示星期一，6表示星期日）
        week = str(dt_object.weekday())
        return time_minute, week

    @staticmethod
    def get_time_frame_dict(start_time, end_time, repeat):
        """
        获取时间范围字典
        @param start_time: 开始时间
        @param end_time: 结束时间
        @param repeat: 星期周期的十进制数,如127 -> 0,1,2,3,4,5,6
        @return: time_frame_dict, time_frame_next_day_dict
        """
        # 十进制转为7位的二进制并倒序
        bin_str = bin(repeat)[2:].zfill(7)[::-1]
        # 生成星期周期列表
        week_list = []
        for i, bit in enumerate(bin_str):
            if bit == '1':
                week_list.append(i)

        # 生成时间范围字典
        time_frame_dict = {}
        time_frame_next_day_dict = {}
        # 非隔天
        if end_time > start_time:
            for week in week_list:
                time_frame_dict[str(week)] = {
                    'start_time': start_time,
                    'end_time': end_time
                }
        # 隔天
        else:
            # time_frame_dict记录当天时间范围,time_frame_next_day_dict记录溢出到第二天的时间范围
            for week in week_list:
                time_frame_dict[str(week)] = {
                    'start_time': start_time,
                    'end_time': 1439    # 23:59
                }
                week += 1
                if week == 7:
                    week = 0
                time_frame_next_day_dict[str(week)] = {
                    'start_time': 0,    # 00:00
                    'end_time': end_time
                }
        return time_frame_dict, time_frame_next_day_dict

    @staticmethod
    def pub_mqtt(device_type, event_type, serial_number, delay_time, scene_id=0):
        """
        发布mqtt消息
        @param device_type: 设备类型
        @param event_type: 事件类型
        @param serial_number: 序列号
        @param delay_time: 延迟时间
        @param scene_id: 场景id
        @return:
        """
        if device_type == DEVICE_TYPE['socket']:
            topic_name = SMART_SOCKET_TOPIC.format(serial_number)
            status = 1 if event_type == EVENT_TYPE['socket_power_on'] else 0
            msg = {
                'type': 1,
                'data': {'deviceSwitch': status}
            }
            if delay_time:
                time.sleep(delay_time)
            CommonService.req_publish_mqtt_msg(serial_number, topic_name, msg)

        # 没有设备任务时,最后一个任务上报场景日志
        if scene_id:
            data = {
                'sceneId': scene_id,
                'status': 1
            }
            url = DETECT_PUSH_DOMAIN + 'gatewayService/sceneLogPush'
            requests.post(url=url, data=data, timeout=8)