ASJServer/Controller/Cron/CronTaskController.py

#!/usr/bin/python3.6
# -*- coding: utf-8 -*-
#
# Copyright (C) 2022 #
# @Time    : 2022/4/1 11:27
# @Author  : ming
# @Email   : zhangdongming@asj6.wecom.work
# @File    : CronTaskController.py
# @Software: PyCharm
import datetime
import logging
import time

import requests
from django.db import connection, connections, transaction
from django.db.models import Q, Sum, Count
from django.views import View

from Ansjer.config import USED_SERIAL_REDIS_LIST, UNUSED_SERIAL_REDIS_LIST, CONFIG_INFO, CONFIG_US, \
    RESET_REGION_ID_SERIAL_REDIS_LIST
from Model.models import Device_User, Device_Info, UidSetModel, UID_Bucket, Unused_Uid_Meal, Order_Model, StsCrdModel, \
    VodHlsModel, ExperienceContextModel, AiService, VodHlsSummary, VideoPlaybackTimeModel, DeviceUserSummary, \
    CountryModel, DeviceTypeModel, OrdersSummary, DeviceInfoSummary, CompanySerialModel, \
    CloudLogModel, UidCloudStorageCount, UserExModel, DeviceDomainRegionModel, VodHlsTag, VodHlsTagType, \
    Store_Meal, Lang, VodBucketModel
from Object.RedisObject import RedisObject
from Object.ResponseObject import ResponseObject
from Object.utils import LocalDateTimeUtil
from Service.CommonService import CommonService
from Service.VodHlsService import SplitVodHlsObject

LOGGER = logging.getLogger('info')


class CronDelDataView(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):
        response = ResponseObject()
        if operation == 'delAccessLog':  # 定时删除访问接口数据
            return self.delAccessLog(response)
        elif operation == 'delPushInfo':  # 定时删除推送数据
            return self.delPushInfo(response)
        elif operation == 'delVodHls':  # 定时删除云存播放列表
            return self.delVodHls(response)
        elif operation == 'delCloudLog':  # 定时删除云存接口数据
            return self.delCloudLog(response)
        elif operation == 'delTesterDevice':  # 定时删除测试账号下的设备数据
            return self.delTesterDevice(response)
        elif operation == 'delAppLog':  # 定时删除app日志
            return self.delAppLog(response)
        elif operation == 'UpdateConfiguration':  # 定时更新配置
            return self.UpdateConfiguration(response)
        elif operation == 'cloud-log':
            return self.uid_cloud_storage_upload_count(response)
        else:
            return response.json(404)

    @staticmethod
    def UpdateConfiguration(response):
        """
        定时更新配置
        @param response: 响应对象
        @return:
        """
        try:
            ucode_list = ['823C01552AA',
                          '823C01550AA',
                          '823C01850XA',
                          '730201350AA',
                          '730201350AA',
                          '730201450AA',
                          '730201450MA',
                          '72V201252AA',
                          '72V201253AA',
                          '72V201353AA',
                          '72V201354AA',
                          '72V201355AA',
                          '72V201254AA']
            UidSetModel.objects.filter(ucode__in=ucode_list, is_human=0).update(is_human=1)
            UidSetModel.objects.filter(ucode='72V201254AA', mobile_4g=0).update(mobile_4g=1)
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def delAppLog(response):
        """
        定时删除app日志
        @param response: 响应对象
        @return:
        """
        nowTime = int(time.time())
        try:
            cursor = connection.cursor()
            month_ago_time = nowTime - 30 * 24 * 60 * 60  # 保留近30天的数据
            sql = 'DELETE FROM `app_log` WHERE add_time<{}'.format(month_ago_time)
            cursor.execute(sql)
            cursor.close()
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def uid_cloud_storage_upload_count(response):
        try:
            now_time = int(time.time())
            local_time = LocalDateTimeUtil.get_before_days_timestamp(now_time)
            format_str = '%Y-%m-%d'
            date_str = LocalDateTimeUtil.time_stamp_to_time(local_time, format_str)
            start_time, end_time = LocalDateTimeUtil.get_start_and_end_time(date_str, format_str)
            cs_uid_qs = UID_Bucket.objects.filter(addTime__gte=int(1669824000)).values('uid')
            if not cs_uid_qs.exists():
                return response.json(0)
            for item in cs_uid_qs:
                uid = item['uid']
                cloud_log_qs = CloudLogModel.objects.filter(uid=uid, operation=r'cloudstorage/storeplaylist',
                                                            time__gte=start_time, time__lte=end_time)
                cloud_log_qs = cloud_log_qs.values('uid')[0:1]
                if not cloud_log_qs.exists():
                    continue
                count_data = {'uid': uid, 'count': cloud_log_qs.count(), 'created_time': end_time,
                              'updated_time': end_time}
                UidCloudStorageCount.objects.create(**count_data)
            return response.json(0)
        except Exception as e:
            LOGGER.info('异常详情,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
            return response.json(500)

    @staticmethod
    def delAccessLog(response):
        try:
            cursor = connection.cursor()
            # 删除7天前的数据
            last_week = LocalDateTimeUtil.get_last_week()
            sql = 'DELETE FROM access_log WHERE time < %s limit %s'
            cursor.execute(sql, [last_week, 10000])
            # 关闭游标
            cursor.close()
            connection.close()
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def delPushInfo(response):
        now_time = int(time.time())
        cursor = connections['mysql02'].cursor()
        try:
            # 当前时间转日期
            local_date_now = str(datetime.datetime.fromtimestamp(int(now_time)).date())
            # 根据日期获取周几
            week_val = LocalDateTimeUtil.date_to_week(local_date_now)
            # 根据当前时间获取7天前时间戳
            expiration_time = LocalDateTimeUtil.get_before_days_timestamp(now_time, 7)
            # 每次删除条数
            size = 10000
            # 删除7天前的数据
            sql = "DELETE FROM equipment_info WHERE addTime<= %s LIMIT %s "
            for i in range(6):
                cursor.execute(sql, [expiration_time, size])
            if week_val == 1:
                sql = "DELETE FROM equipment_info_sunday WHERE add_time<= %s LIMIT %s "
            if week_val == 2:
                sql = "DELETE FROM equipment_info_monday WHERE add_time<= %s LIMIT %s "
            if week_val == 3:
                sql = "DELETE FROM equipment_info_tuesday WHERE add_time<= %s LIMIT %s "
            if week_val == 4:
                sql = "DELETE FROM equipment_info_wednesday WHERE add_time<= %s LIMIT %s "
            if week_val == 5:
                sql = "DELETE FROM equipment_info_thursday WHERE add_time<= %s LIMIT %s "
            if week_val == 6:
                sql = "DELETE FROM equipment_info_friday WHERE add_time<= %s LIMIT %s "
            if week_val == 7:
                sql = "DELETE FROM equipment_info_saturday WHERE add_time<= %s LIMIT %s "
            for i in range(5):
                cursor.execute(sql, [expiration_time, size])
            # 关闭游标
            cursor.close()
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def delVodHls(response):
        nowTime = int(time.time())
        try:
            CronDelDataView.del_vod_hls_tag()
            cursor = connection.cursor()
            month_ago_time = nowTime - 3 * 30 * 24 * 60 * 60  # 删除3个月前的数据
            sql = 'DELETE FROM `vod_hls` WHERE endTime<{} LIMIT 50000'.format(month_ago_time)
            cursor.execute(sql)
            cursor.close()
            # 删除vod_hls分表数据
            split_vod_hls_obj = SplitVodHlsObject()
            split_vod_hls_obj.del_vod_hls_data(end_time__lt=month_ago_time)
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def del_vod_hls_tag():
        """
        删除AI标签记录
        """
        e_time = LocalDateTimeUtil.get_before_days_timestamp(int(time.time()), 30)
        VodHlsTagType.objects.filter(created_time__lt=e_time).delete()
        VodHlsTag.objects.filter(created_time__lt=e_time).delete()

    @staticmethod
    def delCloudLog(response):
        nowTime = int(time.time())
        cursor = connection.cursor()
        try:
            # 删除3个月前的数据
            sql = "DELETE FROM `cloud_log` WHERE time<={} LIMIT 50000".format(
                nowTime - 3 * 30 * 24 * 60 * 60)
            cursor.execute(sql)
            # 关闭游标
            cursor.close()
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def delTesterDevice(response):
        try:
            userID_list = [
                'tech01@ansjer.com',
                'tech02@ansjer.com',
                'tech03@ansjer.com',
                'tech04@ansjer.com',
                'tech05@ansjer.com',
                'tech06@ansjer.com',
                'tech07@ansjer.com',
                'tech08@ansjer.com',
                'tech09@ansjer.com',
                'tech10@ansjer.com',
                'fix01@ansjer.com',
                'fix02@ansjer.com',
                'fix03@ansjer.com',
                'fix04@ansjer.com',
                'fix05@ansjer.com']
            device_user = Device_User.objects.filter(username__in=userID_list)
            device_info_qs = Device_Info.objects.filter(
                userID__in=device_user).values('UID')
            uid_list = []
            for device_info in device_info_qs:
                uid_list.append(device_info['UID'])
            with transaction.atomic():
                # 删除设备云存相关数据
                UidSetModel.objects.filter(uid__in=uid_list).delete()
                UID_Bucket.objects.filter(uid__in=uid_list).delete()
                Unused_Uid_Meal.objects.filter(uid__in=uid_list).delete()
                Order_Model.objects.filter(UID__in=uid_list).delete()
                StsCrdModel.objects.filter(uid__in=uid_list).delete()
                VodHlsModel.objects.filter(uid__in=uid_list).delete()
                # 删除vod_hls分表数据
                split_vod_hls_obj = SplitVodHlsObject()
                split_vod_hls_obj.del_vod_hls_data(uid__in=uid_list)
                ExperienceContextModel.objects.filter(uid__in=uid_list).delete()
                Device_Info.objects.filter(userID__in=device_user).delete()
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))


class CronUpdateDataView(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):
        response = ResponseObject()
        if operation == 'updateUnusedUidBucket':  # 定时更新过期云存关联的未使用套餐状态
            return self.updateUnusedUidBucket(response)
        elif operation == 'updateUnusedAiService':  # 定时更新过期ai关联的未使用套餐状态
            return self.updateUnusedAiService(response)
        elif operation == 'reqUpdateSerialStatus':  # 定时请求更新序列号状态
            return self.reqUpdateSerialStatus(response)
        elif operation == 'updateSerialStatus':  # 更新序列号状态
            return self.updateSerialStatus(request_dict, response)
        elif operation == 'reset-region-id':  # 重置地区id
            return self.reset_region_id(request_dict, response)
        elif operation == 'updateExperienceMeal':  # 定时修改体验套餐有效期为1个月
            return self.update_experience_meal(request_dict, response)
        else:
            return response.json(404)

    @staticmethod
    def updateUnusedUidBucket(response):
        """
        监控云存套餐过期修改状态
        @param response:
        @return:
        """
        # 定时更新已过期套餐修改状态为2
        now_time = int(time.time())
        expired_uid_bucket = UID_Bucket.objects.filter(endTime__lte=now_time)
        expired_uid_bucket = expired_uid_bucket.filter(~Q(use_status=2)).values('id')
        if expired_uid_bucket.exists():
            expired_uid_bucket.update(use_status=2)
        # 监控有未使用套餐则自动生效
        expired_uid_buckets = \
            UID_Bucket.objects.filter(endTime__lte=now_time, has_unused=1).values("id", "uid")[0:1000]
        for expired_uid_bucket in expired_uid_buckets:
            unuseds = Unused_Uid_Meal.objects.filter(
                uid=expired_uid_bucket['uid']).values(
                "id",
                "uid",
                "channel",
                "addTime",
                "expire",
                "num",
                "bucket_id").order_by('addTime')[0:1]
            if not unuseds.exists():
                continue
            unused = unuseds[0]
            try:
                with transaction.atomic():
                    count_unused = Unused_Uid_Meal.objects.filter(
                        uid=expired_uid_bucket['uid']).count()
                    has_unused = 1 if count_unused > 1 else 0
                    endTime = CommonService.calcMonthLater(
                        unused['expire'] * unused['num'])
                    UID_Bucket.objects.filter(
                        uid=expired_uid_bucket['uid']).update(
                        channel=unused['channel'],
                        endTime=endTime,
                        bucket_id=unused['bucket_id'],
                        updateTime=now_time,
                        use_status=1,
                        has_unused=has_unused)
                    Unused_Uid_Meal.objects.filter(id=unused['id']).delete()
                    StsCrdModel.objects.filter(
                        uid=expired_uid_bucket['uid']).delete()  # 删除sts记录
            except Exception as e:
                print(repr(e))
                continue
        return response.json(0)

    @staticmethod
    def updateUnusedAiService(response):
        now_time = int(time.time())
        ai_service_qs = AiService.objects.filter(
            endTime__lte=now_time,
            use_status=1).values(
            'id',
            'uid')[
                        0:200]
        for ai_service in ai_service_qs:
            try:
                with transaction.atomic():
                    AiService.objects.filter(
                        id=ai_service['id']).update(
                        use_status=2)  # 更新过期ai订单状态
                    # 如果存在未使用套餐,更新为使用
                    unused_ai_service = AiService.objects.filter(
                        uid=ai_service['uid'],
                        use_status=0).order_by('addTime')[
                                        :1].values(
                        'id',
                        'endTime')
                    if unused_ai_service.exists():
                        # 未使用套餐的endTime在购买的时候保存为有效时间
                        effective_day = unused_ai_service[0]['endTime']
                        endTime = now_time + effective_day
                        AiService.objects.filter(
                            id=unused_ai_service[0]['id']).update(
                            use_status=1, endTime=endTime, updTime=now_time)
            except Exception:
                continue
        return response.json(0)

    @classmethod
    def reqUpdateSerialStatus(cls, response):
        redis_obj = RedisObject()
        logger = logging.getLogger('info')

        # 更新已使用序列号其他服务器的状态
        used_serial_redis_list = redis_obj.lrange(USED_SERIAL_REDIS_LIST, 0, -1)  # 读取redis已使用序列号
        if used_serial_redis_list:
            logger.info('---请求更新已使用序列号列表---used_serial_redis_list:{}'.format(used_serial_redis_list))
            used_serial_redis_list = [str(i, 'utf-8') for i in used_serial_redis_list]
            cls.do_request_function(used_serial_redis_list, 3)

        # 更新未使用序列号其他服务器的状态
        unused_serial_redis_list = redis_obj.lrange(UNUSED_SERIAL_REDIS_LIST, 0, -1)  # 读取redis未使用序列号
        if unused_serial_redis_list:
            logger.info('---请求更新未使用序列号列表---unused_serial_redis_list:{}'.format(unused_serial_redis_list))
            unused_serial_redis_list = [str(i, 'utf-8') for i in unused_serial_redis_list]
            cls.do_request_function(unused_serial_redis_list, 1)

        # 重置地区id
        reset_region_id_serial_redis_list = redis_obj.lrange(RESET_REGION_ID_SERIAL_REDIS_LIST, 0, -1)  # 读取redis未使用序列号
        if reset_region_id_serial_redis_list:
            logger.info('---请求重置地区id的序列号列表---:{}'.format(reset_region_id_serial_redis_list))
            reset_region_id_serial_redis_list = [str(i, 'utf-8') for i in reset_region_id_serial_redis_list]
            cls.do_request_reset_region_id(reset_region_id_serial_redis_list)
        return response.json(0)

    @staticmethod
    def do_request_function(serial_redis_list, status):
        """
        请求更新序列号状态
        @param serial_redis_list: 序列号redis列表
        @param status: 状态, 1: 未使用, 3: 已占用
        """
        data = {
            'serial_redis_list': str(serial_redis_list),
            'status': status
        }
        # 确认域名列表
        orders_domain_name_list = CommonService.get_orders_domain_name_list()
        redis_obj = RedisObject()
        logger = logging.getLogger('info')
        logger.info('---请求更新序列号线程---data:{},orders_domain_name_list:{}'.format(data, orders_domain_name_list))
        try:
            requests_failed_flag = False  # 请求失败标志位
            for domain_name in orders_domain_name_list:
                url = '{}cron/update/updateSerialStatus'.format(domain_name)
                response = requests.post(url=url, data=data, timeout=5)
                logger.info('---请求更新序列号响应时间---:{}'.format(response.elapsed.total_seconds()))
                result = response.json()
                if result['result_code'] != 0:  # 请求失败标志位置位
                    requests_failed_flag = True
                    break

                # 状态为未使用,重置美洲服的地区id
                if status == 1:     # 美洲服直接更新
                    if CONFIG_INFO == CONFIG_US:
                        DeviceDomainRegionModel.objects.filter(~Q(region_id=0), serial_number__in=serial_redis_list).\
                            update(region_id=0)
                    else:   # 其他服请求到美洲服更新
                        req_url = 'https://www.dvema.com/cron/update/reset-region-id'
                        req_data = {
                            'serial_redis_list': str(serial_redis_list)
                        }
                        response = requests.post(url=req_url, data=req_data, timeout=5)
                        logger.info('---请求重置地区id响应时间---:{}'.format(response.elapsed.total_seconds()))
                        result = response.json()
                        if result['result_code'] != 0:  # 请求失败标志位置位
                            requests_failed_flag = True
                            break

            if not requests_failed_flag:  # 请求成功删除redis序列号
                if status == 1:
                    for i in serial_redis_list:
                        redis_obj.lrem(UNUSED_SERIAL_REDIS_LIST, 0, i)
                elif status == 3:
                    for i in serial_redis_list:
                        redis_obj.lrem(USED_SERIAL_REDIS_LIST, 0, i)
        except Exception as e:
            logger.info('---更新序列号状态异常---:{}'.format(repr(e)))

    @staticmethod
    def do_request_reset_region_id(reset_region_id_serial_redis_list):
        """
        请求重置地区id
        @param reset_region_id_serial_redis_list: 序列号redis列表
        """
        redis_obj = RedisObject()
        logger = logging.getLogger('info')
        requests_failed_flag = False  # 请求失败标志位
        data = {
            'serial_redis_list': str(reset_region_id_serial_redis_list),
        }
        url = 'https://www.dvema.com/cron/update/reset-region-id'

        try:
            response = requests.post(url=url, data=data, timeout=5)
            result = response.json()
            if result['result_code'] != 0:  # 请求失败标志位置位
                requests_failed_flag = True
            if not requests_failed_flag:  # 请求成功删除redis序列号
                for serial in reset_region_id_serial_redis_list:
                    redis_obj.lrem(RESET_REGION_ID_SERIAL_REDIS_LIST, 0, serial)
        except Exception as e:
            logger.info('---请求重置地区id异常---:{}'.format(repr(e)))

    @staticmethod
    def updateSerialStatus(request_dict, response):
        """
        更新序列号状态
        @param request_dict: 请求参数
        @request_dict serial_redis_list: 序列号redis列表
        @request_dict status: 状态, 1: 未使用, 3: 已占用
        @param response: 响应对象
        """
        serial_redis_list = request_dict.get('serial_redis_list', None)
        status = request_dict.get('status', None)
        logger = logging.getLogger('info')
        logger.info('---更新序列号状态参数---serial_redis_list:{},status:{}'.format(serial_redis_list, status))
        if not all([serial_redis_list, status]):
            return response.json(444)
        now_time = int(time.time())
        try:
            serial_redis_list = eval(serial_redis_list)
            CompanySerialModel.objects.filter(serial_number__in=serial_redis_list).update(status=int(status),
                                                                                          update_time=now_time)
            return response.json(0)
        except Exception as e:
            logger.info('---更新序列号状态异常---:{}'.format(repr(e)))
            return response.json(500)

    @staticmethod
    def reset_region_id(request_dict, response):
        """
        重置地区id
        @param request_dict: 请求参数
        @request_dict serial_redis_list: 序列号redis列表
        @param response: 响应对象
        """
        serial_redis_list = request_dict.get('serial_redis_list', None)
        logger = logging.getLogger('info')
        logger.info('---重置地区id参数---serial_redis_list:{}'.format(serial_redis_list))
        if not serial_redis_list:
            return response.json(444)
        try:
            serial_redis_list = eval(serial_redis_list)
            DeviceDomainRegionModel.objects.filter(serial_number__in=serial_redis_list).update(region_id=0)
            return response.json(0)
        except Exception as e:
            logger.info('---重置地区id异常---:{}'.format(repr(e)))
            return response.json(500)

    @staticmethod
    def update_experience_meal(request_dict, response):
        """
        定时修改体验套餐有效期为1个月
        @param request_dict: 请求参数
        @param response: 响应对象
        """
        try:
            meal_qs = Store_Meal.objects.filter(is_show=0, pay_type=10, expire=3, day=7).values('id', 'bucket')
            meal_id = meal_qs[0]['id']
            lang_qs = Lang.objects.filter(store_meal__id=meal_id).values('lang')
            for item in lang_qs:
                lang = item['lang']
                if lang == 'cn':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='一个月免费套餐')
                elif lang == 'en':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='1-Month plan (free trial)')
                elif lang == 'es':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='Plan de 1 mes (prueba gratuita)')
                elif lang == 'fr':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='Forfait de 1 mois (essai gratuit)')
                elif lang == 'de':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='1 Monat Paket (kostenlose Testversion) ')
                elif lang == 'cn_tw':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='一個月套餐（免費試用）')
                elif lang == 'pt':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='Plano de 1 mês (teste gratuito)')
                elif lang == 'ru':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='Тариф 1 месяц (бесплатный пробный период)')
                elif lang == 'ja':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='1ヶ月プラン（無料試用）')
                elif lang == 'it':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='Pacchetto di 1 mese (prova gratuita)')
                elif lang == 'pl':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='jednomiesięczny pakiet (bezpłatny próbny)')
                elif lang == 'nl':
                    Lang.objects.filter(lang=lang, store_meal__id=meal_id).update(content='Pakket van 1 maand (gratis proefperiode)')
            VodBucketModel.objects.filter(id=meal_qs[0]['bucket']).update(content='国内存储桶免费体验30天，录像保存7天')
            meal_qs.update(expire=1)
            return response.json(0)
        except Exception as e:
            LOGGER.info('---修改体验套餐有效期---:{}'.format(repr(e)))
            return response.json(500)

class CronCollectDataView(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):
        response = ResponseObject()
        if operation == 'collectPlayBack':  # 定时保存云存视频回放
            return self.collect_play_back(response)
        elif operation == 'collectDeviceUser':  # 定时保存用户数据
            return self.collect_device_user(response)
        elif operation == 'collectOrder':  # 定时保存订单数据
            return self.collect_order(response)
        elif operation == 'collectDeviceInfo':  # 定时保存设备数据
            return self.collect_device_info(response)
        else:
            return response.json(404)

    @staticmethod
    def collect_play_back(response):
        try:
            now_time = int(time.time())
            today = datetime.datetime.today()
            start_time = datetime.datetime(today.year, today.month, today.day)
            end_time = start_time + datetime.timedelta(days=1)
            start_time = CommonService.str_to_timestamp(start_time.strftime('%Y-%m-%d %H:%M:%S'))
            end_time = CommonService.str_to_timestamp(end_time.strftime('%Y-%m-%d %H:%M:%S'))
            this_month_str = datetime.datetime(today.year, today.month, 1)
            this_month_stamp = CommonService.str_to_timestamp(this_month_str.strftime('%Y-%m-%d %H:%M:%S'))
            video_play_back_time_qs = VideoPlaybackTimeModel.objects.filter(startTime__gte=start_time,
                                                                            startTime__lt=end_time,
                                                                            playMode='cloud').values('uid').annotate(
                play_duration=Sum('duration'), play_frequency=Count('uid'))
            with transaction.atomic():
                for item in video_play_back_time_qs:
                    vod_hls_summary_qs = VodHlsSummary.objects.filter(uid=item['uid'], time=this_month_stamp)
                    if vod_hls_summary_qs.exists():
                        vod_hls_summary = vod_hls_summary_qs.first()
                        vod_hls_summary.play_duration += item['play_duration']
                        vod_hls_summary.play_frequency += 1
                        vod_hls_summary.updated_time = now_time
                        vod_hls_summary.save()
                    else:
                        VodHlsSummary.objects.create(uid=item['uid'], time=this_month_stamp, created_time=now_time,
                                                     play_duration=item['play_duration'], play_frequency=1,
                                                     updated_time=now_time)

            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def collect_device_user(response):
        try:
            created_time = int(time.time())
            today = datetime.datetime.today()
            start_time = datetime.datetime(today.year, today.month, today.day)
            end_time = start_time + datetime.timedelta(days=1)
            increase_user_qs = Device_User.objects.filter(data_joined__year=today.year, data_joined__month=today.month,
                                                          data_joined__day=today.day).values('region_country')
            start_time = CommonService.str_to_timestamp(start_time.strftime('%Y-%m-%d %H:%M:%S'))
            end_time = CommonService.str_to_timestamp(end_time.strftime('%Y-%m-%d %H:%M:%S'))
            active_user_qs = UserExModel.objects.filter(updTime__gte=start_time, updTime__lt=end_time).values(
                'userID__region_country')
            country_qs = CountryModel.objects.all().values('id', 'region__name', 'country_name')
            country_dict = {}
            continent_dict = {}
            for item in country_qs:
                country_dict[item['id']] = item['country_name']
                continent_dict[item['country_name']] = item['region__name']
            with transaction.atomic():
                if increase_user_qs.exists():
                    increase_user_count = increase_user_qs.count()
                    increase_user_country_list = increase_user_qs.values('region_country').annotate(
                        count=Count('region_country')).order_by('count')
                    increase_user_country_dict = {}
                    increase_user_continent_dict = {}
                    for item in increase_user_country_list:
                        country_name = country_dict.get(item['region_country'], '未知国家')
                        continent_name = continent_dict.get(country_name, '未知大洲')
                        increase_user_country_dict[country_name] = item['count']
                        if continent_name not in increase_user_continent_dict:
                            increase_user_continent_dict[continent_name] = 0
                        increase_user_continent_dict[continent_name] += item['count']
                    DeviceUserSummary.objects.create(time=start_time, count=increase_user_count,
                                                     country=increase_user_country_dict, created_time=created_time,
                                                     continent=increase_user_continent_dict)
                if active_user_qs.exists():
                    active_user_count = active_user_qs.count()
                    active_user_country_list = active_user_qs.values('userID__region_country').annotate(
                        count=Count('userID__region_country')).order_by('count')
                    active_user_country_dict = {}
                    active_user_continent_dict = {}
                    for item in active_user_country_list:
                        country_name = country_dict.get(item['userID__region_country'], '未知国家')
                        continent_name = continent_dict.get(country_name, '未知大洲')
                        active_user_country_dict[country_name] = item['count']
                        if continent_name not in active_user_continent_dict:
                            active_user_continent_dict[continent_name] = 0
                        active_user_continent_dict[continent_name] += item['count']
                    DeviceUserSummary.objects.create(time=start_time, query_type=1, count=active_user_count,
                                                     country=active_user_country_dict, created_time=created_time,
                                                     continent=active_user_continent_dict)

            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def collect_order(response):
        try:
            created_time = int(time.time())
            today = datetime.datetime.today()
            start_time = datetime.datetime(today.year, today.month, today.day)
            end_time = start_time + datetime.timedelta(days=1)
            start_time = CommonService.str_to_timestamp(start_time.strftime('%Y-%m-%d %H:%M:%S'))
            end_time = CommonService.str_to_timestamp(end_time.strftime('%Y-%m-%d %H:%M:%S'))
            order_qs = Order_Model.objects.filter(addTime__gte=start_time, addTime__lt=end_time,
                                                  status=1).values('UID', 'order_type',
                                                                   'store_meal_name', 'price',
                                                                   'addTime', 'currency').order_by(
                'addTime')
            uid_list = []
            all_order_qs = Order_Model.objects.filter(addTime__lt=start_time, status=1).values('UID')
            for item in all_order_qs:
                if item['UID'] not in uid_list:
                    uid_list.append(item['UID'])
            # 国家表数据
            country_qs = CountryModel.objects.values('id', 'country_name')
            country_dict = {}
            for item in country_qs:
                country_dict[item['id']] = item['country_name']
            # 设备类型数据
            device_type_qs = DeviceTypeModel.objects.values('name', 'type')
            device_type_dict = {}
            for item in device_type_qs:
                device_type_dict[item['type']] = item['name']

            with transaction.atomic():
                for item in order_qs:
                    is_pay = 0
                    price = float(item['price'])
                    currency = item['currency']
                    uid_set_qs = UidSetModel.objects.filter(uid=item['UID']).values('tb_country')
                    country_id = uid_set_qs[0]['tb_country'] if uid_set_qs.exists() else 0
                    country_name = country_dict.get(country_id, '未知国家')
                    order_type = item['order_type']
                    device_info_qs = Device_Info.objects.filter(UID=item['UID']).values('Type')
                    device_type_id = device_info_qs[0]['Type'] if device_info_qs.exists() else 0
                    device_type_name = device_type_dict.get(device_type_id, '未知设备')
                    store_meal_name = item['store_meal_name']
                    add_time_stamp = item['addTime']
                    add_time_str = datetime.datetime.fromtimestamp(int(add_time_stamp))
                    add_time_str = datetime.datetime(add_time_str.year, add_time_str.month, add_time_str.day)
                    add_time_stamp = CommonService.str_to_timestamp(add_time_str.strftime('%Y-%m-%d %H:%M:%S'))
                    if price == 0:
                        is_pay = 1
                        order_summary_qs = OrdersSummary.objects.filter(time=add_time_stamp, query_type=1,
                                                                        service_type=order_type)
                    else:
                        order_summary_qs = OrdersSummary.objects.filter(time=add_time_stamp, query_type=0,
                                                                        service_type=order_type)
                        if item['UID'] not in uid_list:
                            pay_order_summary_qs = OrdersSummary.objects.filter(time=add_time_stamp, query_type=2,
                                                                                service_type=order_type)
                            query_type = 2
                        else:
                            pay_order_summary_qs = OrdersSummary.objects.filter(time=add_time_stamp, query_type=3,
                                                                                service_type=order_type)
                            query_type = 3
                        if pay_order_summary_qs.exists():
                            pay_order_summary = pay_order_summary_qs.first()
                            pay_order_summary.count += 1
                            temp_total = eval(pay_order_summary.total)
                            if currency not in temp_total:
                                temp_total[currency] = price
                            else:
                                temp_total[currency] = round(temp_total[currency] + price, 2)
                            pay_order_summary.total = temp_total
                            country_temp_dict = eval(pay_order_summary.country)
                            if country_name in country_temp_dict:
                                country_temp_dict[country_name]['数量'] += 1
                                if currency not in country_temp_dict[country_name]:
                                    country_temp_dict[country_name][currency] = price
                                else:
                                    country_temp_dict[country_name][currency] = round(
                                        country_temp_dict[country_name][currency] + price, 2)
                            else:
                                country_temp_dict[country_name] = {'数量': 1, currency: price}
                            pay_order_summary.country = country_temp_dict
                            device_type_temp_dict = eval(pay_order_summary.device_type)
                            if device_type_name in device_type_temp_dict:
                                device_type_temp_dict[device_type_name]['数量'] += 1
                                if currency not in device_type_temp_dict[device_type_name]:
                                    device_type_temp_dict[device_type_name][currency] = price
                                else:
                                    device_type_temp_dict[device_type_name][currency] = round(
                                        device_type_temp_dict[device_type_name][currency] + price, 2)
                            else:
                                device_type_temp_dict[device_type_name] = {'数量': 1, currency: price}
                            pay_order_summary.device_type = device_type_temp_dict
                            store_meal_temp_dict = eval(pay_order_summary.store_meal)
                            if store_meal_name in store_meal_temp_dict:
                                store_meal_temp_dict[store_meal_name]['数量'] += 1
                                if currency not in store_meal_temp_dict[store_meal_name]:
                                    store_meal_temp_dict[store_meal_name][currency] = price
                                else:
                                    store_meal_temp_dict[store_meal_name][currency] = round(
                                        store_meal_temp_dict[store_meal_name][currency] + price, 2)
                            else:
                                store_meal_temp_dict[store_meal_name] = {'数量': 1, currency: price}
                            pay_order_summary.store_meal = store_meal_temp_dict
                            pay_order_summary.save()
                        else:
                            final_total = {currency: price}
                            country_temp_dict = {
                                country_name: {
                                    '数量': 1,
                                    currency: price
                                }
                            }
                            device_type_temp_dict = {
                                device_type_name: {
                                    '数量': 1,
                                    currency: price
                                }
                            }
                            store_meal_temp_dict = {
                                store_meal_name: {
                                    '数量': 1,
                                    currency: price
                                }
                            }
                            OrdersSummary.objects.create(time=add_time_stamp, count=1, query_type=query_type,
                                                         service_type=order_type, total=final_total,
                                                         country=country_temp_dict, created_time=created_time,
                                                         device_type=device_type_temp_dict,
                                                         store_meal=store_meal_temp_dict)
                    if order_summary_qs.exists():
                        order_summary = order_summary_qs.first()
                        order_summary.count += 1
                        temp_total = eval(order_summary.total)
                        if currency not in temp_total:
                            temp_total[currency] = price
                        else:
                            temp_total[currency] = round(temp_total[currency] + price, 2)
                        order_summary.total = temp_total
                        country_temp_dict = eval(order_summary.country)
                        if country_name in country_temp_dict:
                            if is_pay == 0:
                                country_temp_dict[country_name]['数量'] += 1
                                if currency not in country_temp_dict[country_name]:
                                    country_temp_dict[country_name][currency] = price
                                else:
                                    country_temp_dict[country_name][currency] = round(
                                        country_temp_dict[country_name][currency] + price, 2)
                            else:
                                country_temp_dict[country_name] += 1
                        else:
                            if is_pay == 0:
                                country_temp_dict[country_name] = {'数量': 1, currency: price}
                            else:
                                country_temp_dict[country_name] = 1
                        order_summary.country = country_temp_dict
                        device_type_temp_dict = eval(order_summary.device_type)
                        if device_type_name in device_type_temp_dict:
                            if is_pay == 0:
                                device_type_temp_dict[device_type_name]['数量'] += 1
                                if currency not in device_type_temp_dict[device_type_name]:
                                    device_type_temp_dict[device_type_name][currency] = price
                                else:

                                    device_type_temp_dict[device_type_name][currency] = round(
                                        device_type_temp_dict[device_type_name][currency] + price, 2)
                            else:
                                device_type_temp_dict[device_type_name] += 1
                        else:
                            if is_pay == 0:
                                device_type_temp_dict[device_type_name] = {'数量': 1, currency: price}
                            else:
                                device_type_temp_dict[device_type_name] = 1
                        order_summary.device_type = device_type_temp_dict
                        store_meal_temp_dict = eval(order_summary.store_meal)
                        if store_meal_name in store_meal_temp_dict:
                            if is_pay == 0:
                                store_meal_temp_dict[store_meal_name]['数量'] += 1
                                if currency not in store_meal_temp_dict[store_meal_name]:
                                    store_meal_temp_dict[store_meal_name][currency] = price
                                else:
                                    store_meal_temp_dict[store_meal_name][currency] = round(
                                        store_meal_temp_dict[store_meal_name][currency] + price, 2)
                            else:
                                store_meal_temp_dict[store_meal_name] += 1
                        else:
                            if is_pay == 0:
                                store_meal_temp_dict[store_meal_name] = {'数量': 1, currency: price}
                            else:
                                store_meal_temp_dict[store_meal_name] = 1
                        order_summary.store_meal = store_meal_temp_dict
                        order_summary.save()
                    else:
                        final_total = {currency: price}
                        if is_pay == 0:
                            country_temp_dict = {
                                country_name: {
                                    '数量': 1,
                                    currency: price
                                }
                            }
                            device_type_temp_dict = {
                                device_type_name: {
                                    '数量': 1,
                                    currency: price
                                }
                            }
                            store_meal_temp_dict = {
                                store_meal_name: {
                                    '数量': 1,
                                    currency: price
                                }
                            }
                        else:
                            device_type_temp_dict = {
                                device_type_name: 1
                            }
                            store_meal_temp_dict = {
                                store_meal_name: 1
                            }
                            country_temp_dict = {
                                country_name: 1
                            }
                        OrdersSummary.objects.create(time=add_time_stamp, count=1, query_type=is_pay,
                                                     service_type=order_type, total=final_total,
                                                     country=country_temp_dict, created_time=created_time,
                                                     device_type=device_type_temp_dict, store_meal=store_meal_temp_dict)
            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))

    @staticmethod
    def collect_device_info(response):
        try:
            created_time = int(time.time())
            today = datetime.datetime.today()
            start_time = datetime.datetime(today.year, today.month, today.day)
            end_time = start_time + datetime.timedelta(days=1)
            start_time = CommonService.str_to_timestamp(start_time.strftime('%Y-%m-%d %H:%M:%S'))
            end_time = CommonService.str_to_timestamp(end_time.strftime('%Y-%m-%d %H:%M:%S'))
            increase_device_qs = UidSetModel.objects.filter(addTime__gte=start_time, addTime__lt=end_time).values(
                'tb_country',
                'uid',
                'device_type',
                'cloud_vod',
                'is_ai',
                'mobile_4g',
                'addTime')
            video_play_back_time_qs = VideoPlaybackTimeModel.objects.filter(startTime__gte=start_time,
                                                                            startTime__lt=end_time).values('uid')
            active_device_qs = UidSetModel.objects.filter(uid__in=video_play_back_time_qs).values('tb_country',
                                                                                                  'addTime',
                                                                                                  'device_type',
                                                                                                  'cloud_vod',
                                                                                                  'is_ai',
                                                                                                  'mobile_4g',
                                                                                                  'uid')
            increase_device_count = increase_device_qs.count()
            active_device_count = active_device_qs.count()
            # 国家表数据
            country_qs = CountryModel.objects.values('id', 'country_name', 'region__name')
            country_dict = {}
            continent_dict = {}
            for item in country_qs:
                country_dict[item['id']] = item['country_name']
                continent_dict[item['country_name']] = item['region__name']
            # 设备类型数据
            device_type_qs = DeviceTypeModel.objects.values('name', 'type')
            device_type_dict = {}
            for item in device_type_qs:
                device_type_dict[item['type']] = item['name']
            with transaction.atomic():
                if increase_device_qs.exists():
                    # 国家大洲设备数据
                    increase_device_country_list = increase_device_qs.values('tb_country').annotate(
                        count=Count('tb_country')).order_by('count')
                    increase_device_country_dict = {}
                    increase_device_continent_dict = {}
                    for item in increase_device_country_list:
                        country_name = country_dict.get(item['tb_country'], '未知国家')
                        continent_name = continent_dict.get(country_name, '未知大洲')
                        increase_device_country_dict[country_name] = item['count']
                        if continent_name not in increase_device_continent_dict:
                            increase_device_continent_dict[continent_name] = 0
                        increase_device_continent_dict[continent_name] += item['count']
                    # 设备类型数据
                    increase_device_type_list = increase_device_qs.values('device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    increase_device_type_dict = {}
                    for item in increase_device_type_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        increase_device_type_dict[type_name] = item['count']
                    # 云存设备类型数据
                    increase_device_vod_list = increase_device_qs.filter(~Q(cloud_vod=2)).values(
                        'device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    increase_device_vod_dict = {}
                    for item in increase_device_vod_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        increase_device_vod_dict[type_name] = item['count']
                    # AI设备类型数据
                    increase_device_ai_list = increase_device_qs.filter(~Q(is_ai=2)).values('device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    increase_device_ai_dict = {}
                    for item in increase_device_ai_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        increase_device_ai_dict[type_name] = item['count']
                    # 联通设备类型数据
                    increase_device_unicom_list = increase_device_qs.filter(~Q(mobile_4g=2)).values(
                        'device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    increase_device_unicom_dict = {}
                    for item in increase_device_unicom_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        increase_device_unicom_dict[type_name] = item['count']
                    DeviceInfoSummary.objects.create(time=start_time, count=increase_device_count,
                                                     query_type=0, created_time=created_time,
                                                     country=increase_device_country_dict,
                                                     continent=increase_device_continent_dict,
                                                     vod_service=increase_device_vod_dict,
                                                     ai_service=increase_device_ai_dict,
                                                     unicom_service=increase_device_unicom_dict,
                                                     device_type=increase_device_type_dict)
                if active_device_qs.exists():
                    # 国家大洲设备数据
                    active_device_country_list = active_device_qs.values('tb_country').annotate(
                        count=Count('tb_country')).order_by('count')
                    active_device_country_dict = {}
                    active_device_continent_dict = {}
                    for item in active_device_country_list:
                        country_name = country_dict.get(item['tb_country'], '未知国家')
                        continent_name = continent_dict.get(country_name, '未知大洲')
                        active_device_country_dict[country_name] = item['count']
                        if continent_name not in active_device_continent_dict:
                            active_device_continent_dict[continent_name] = 0
                        active_device_continent_dict[continent_name] += item['count']
                    # 设备类型数据
                    active_device_type_list = active_device_qs.values('device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    active_device_type_dict = {}
                    for item in active_device_type_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        active_device_type_dict[type_name] = item['count']
                    # 云存设备类型数据
                    active_device_vod_list = active_device_qs.filter(~Q(cloud_vod=2)).values('device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    active_device_vod_dict = {}
                    for item in active_device_vod_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        active_device_vod_dict[type_name] = item['count']
                    # AI设备类型数据
                    active_device_ai_list = active_device_qs.filter(~Q(is_ai=2)).values('device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    active_device_ai_dict = {}
                    for item in active_device_ai_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        active_device_ai_dict[type_name] = item['count']
                    # 联通设备类型数据
                    active_device_unicom_list = active_device_qs.filter(~Q(mobile_4g=2)).values('device_type').annotate(
                        count=Count('device_type')).order_by('count')
                    active_device_unicom_dict = {}
                    for item in active_device_unicom_list:
                        type_name = device_type_dict.get(item['device_type'], '未知设备类型')
                        active_device_unicom_dict[type_name] = item['count']
                    DeviceInfoSummary.objects.create(time=start_time, count=active_device_count,
                                                     query_type=1, created_time=created_time,
                                                     country=active_device_country_dict,
                                                     continent=active_device_continent_dict,
                                                     vod_service=active_device_vod_dict,
                                                     ai_service=active_device_ai_dict,
                                                     unicom_service=active_device_unicom_dict,
                                                     device_type=active_device_type_dict)

            return response.json(0)
        except Exception as e:
            return response.json(500, repr(e))