ASJPush/Controller/AiController.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@Copyright (C) ansjer cop Video Technology Co.,Ltd.All rights reserved.
@software: PyCharm
@Version: python3.6
@MODIFY DECORD:ansjer dev
"""
import base64
import json
import logging
import os
import threading
import time

import apns2
import boto3
import jpush
from boto3.session import Session
from django.views.generic.base import View
from pyfcm import FCMNotification

from AnsjerPush.config import AI_IDENTIFICATION_TAGS_DICT, CONFIG_US, CONFIG_EUR
from AnsjerPush.config import APNS_MODE, APNS_CONFIG, BASE_DIR, \
    JPUSH_CONFIG, FCM_CONFIG
from AnsjerPush.config import CONFIG_INFO
from AnsjerPush.config import PUSH_BUCKET
from Model.models import UidPushModel, AiService, VodHlsTag, VodHlsTagType
from Object import MergePic
from Object.DynamodbObject import DynamodbObject
from Object.ETkObject import ETkObject
from Object.OCIObjectStorage import OCIObjectStorage
from Object.RedisObject import RedisObject
from Object.ResponseObject import ResponseObject
from Object.SageMakerAiObject import SageMakerAiObject
from Object.TokenObject import TokenObject
from Object.enums.MessageTypeEnum import MessageTypeEnum
from Service.CommonService import CommonService
from Service.DevicePushService import DevicePushService
from Service.EquipmentInfoService import EquipmentInfoService
from Object.NovaImageTagObject import NovaImageTagObject
from datetime import datetime, timedelta
from django.conf import settings
AWS_ACCESS_KEY_ID = settings.AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY = settings.AWS_SECRET_ACCESS_KEY

TIME_LOGGER = logging.getLogger('time')
# 1. 声明一个全局变量，用于存放创建好的实例
nova_image_tag_instance = None

# 2. 根据条件，创建实例并赋值给这个全局变量
# 这段代码会放在文件的最顶部，它只在 Gunicorn worker 进程启动时执行一次
if CONFIG_INFO == CONFIG_EUR:
    nova_image_tag_instance = NovaImageTagObject(
        AWS_ACCESS_KEY_ID[1], AWS_SECRET_ACCESS_KEY[1], 'eu-west-2'
    )
else:
    nova_image_tag_instance = NovaImageTagObject(
        AWS_ACCESS_KEY_ID[1], AWS_SECRET_ACCESS_KEY[1], 'us-east-1'
    )


# AI服务
class AiView(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 is None:
            return response.json(444, 'error path')
        elif operation == 'identification':  # ai识别
            return self.do_ai_identification(request.POST, response)
        elif operation == 'aiPushStats':  # ai调用统计
            return self.ai_push_stats(request_dict, response)
        else:
            token = request_dict.get('token', None)
            # 设备主键uid
            tko = TokenObject(token)
            response.lang = tko.lang
            if tko.code != 0:
                return response.json(tko.code)
            userID = tko.userID
            if operation == 'identification':  # ai识别
                return self.do_ai_identification(request_dict, response)
            else:
                return response.json(414)

    def do_ai_identification(self, request_dict, response):
        etk = request_dict.get('etk', None)
        n_time = request_dict.get('n_time', None)
        channel = request_dict.get('channel', '1')
        receiveTime = int(time.time())
        TIME_LOGGER.info('*****进入into----ai--api,etk={etk}'.format(etk=etk))
        if not etk:
            return response.json(444)
        dir_path = ''
        uid = ''
        try:
            # 解密uid及判断长度
            eto = ETkObject(etk)
            uid = eto.uid
            TIME_LOGGER.info(f'etk解析uid={uid},n_time={n_time},etk:{etk}')
            if len(uid) != 20 and len(uid) != 14:
                return response.json(444)

            # 通过uid查出endTime是否过期，并且ai开关是否打开
            AiServiceQuery = AiService.objects.filter(uid=uid, detect_status=1, use_status=1, endTime__gt=receiveTime) \
                .values('detect_group', 'orders__payType', 'addTime')
            if not AiServiceQuery.exists():
                TIME_LOGGER.info(f'uid={uid}AI服务未开通或已到期')
                return response.json(173)
            detect_group = AiServiceQuery[0]['detect_group']

            file_post_one = request_dict.get('fileOne', None)
            file_post_two = request_dict.get('fileTwo', None)
            file_post_three = request_dict.get('fileThree', None)

            file_list = [file_post_one, file_post_two, file_post_three]
            del file_post_one, file_post_two, file_post_three
            if not all(file_list):
                for k, val in enumerate(file_list):
                    if not val:
                        TIME_LOGGER.error('{}缺少图片{}'.format(uid, k))
                        return response.json(444, '缺少第{k}张图'.format(k=k + 1))
            redis_obj = RedisObject(db=6)
            ai_key = f'PUSH:AI:{uid}:{channel}'
            ai_data = redis_obj.get_data(ai_key)
            if ai_data:
                return response.json(0, {'msg': 'Push again in one minute'})
            # === AI 调用计数===
            n_time_int = int(n_time) if n_time else int(time.time())
            date_str = time.strftime("%Y%m%d", time.localtime(n_time_int))
            self.ai_push_count(uid, date_str)
            # =========================================
            # 查询推送数据
            uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid). \
                values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id',
                       'userID__NickName',
                       'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval',
                       'uid_set__detect_group', 'uid_set__new_detect_interval',
                       'uid_set__channel', 'uid_set__msg_notify')
            if not uid_push_qs.exists():
                TIME_LOGGER.info(f'uid={uid},用户没有开启AI推送')
                return response.json(173)

            ai_server = 'sageMaker'
            if AiServiceQuery[0]['orders__payType'] == 10:  # AI首次体验前半个月调Rekognition
                now_time = int(time.time())
                add_time = AiServiceQuery[0]['addTime']
                if (now_time - add_time) <= (3600 * 24 * 2):
                    ai_server = 'rekognition'
                elif (now_time - add_time) <= (3600 * 24 * 3):
                    ai_server = 'novaLite'

            APP_NOTIFY_KEY = f'ASJ:NOTIFY:PUSH:{uid}:{channel}'  # 推送间隔缓存KEY
            push_cache_data = redis_obj.get_data(APP_NOTIFY_KEY)
            is_push = False if push_cache_data else True

            notify_data = uid_push_qs[0]['uid_set__msg_notify']

            # APP推送提醒状态
            notify = self.is_ai_push(uid, notify_data) if is_push else is_push
            # nova_key = f'PUSH:NOVA:LITE:{uid}'
            # nova = redis_obj.get_data(nova_key)
            if ai_server == 'novaLite':  # AWS AI模型
                sage_maker = SageMakerAiObject()

                # AI nova识别异步存表&推送
                push_thread = threading.Thread(
                    target=self.async_detection_image_label,
                    kwargs={'sage_maker': sage_maker, 'uid': uid, 'n_time': n_time, 'uid_push_qs': uid_push_qs,
                            'channel': channel, 'file_list': file_list, 'notify': notify, 'detect_group': detect_group})
                push_thread.start()

                self.add_push_cache(APP_NOTIFY_KEY, redis_obj, push_cache_data,
                                    uid_push_qs[0]['uid_set__new_detect_interval'])
                redis_obj.set_data(ai_key, uid, 60)
                return response.json(0)
            if ai_server == 'sageMaker':  # 自建模型sageMaker AI
                sage_maker = SageMakerAiObject()
                ai_result = sage_maker.sage_maker_ai_server(uid, file_list)  # 图片base64识别AI标签
                if ai_result:
                    if ai_result == 'imageError':
                        return response.json(0)
                    res = sage_maker.get_table_name(uid, ai_result, detect_group)
                    if not res:  # 当前识别结果未匹配
                        return response.json(0)
                    push_thread = threading.Thread(
                        target=self.async_message_push,
                        kwargs={'sage_maker': sage_maker, 'uid': uid, 'n_time': n_time, 'uid_push_qs': uid_push_qs,
                                'channel': channel, 'res': res, 'file_list': file_list, 'notify': notify})
                    push_thread.start()  # AI识别异步存表&推送
                    self.add_push_cache(APP_NOTIFY_KEY, redis_obj, push_cache_data,
                                        uid_push_qs[0]['uid_set__new_detect_interval'])
                    redis_obj.set_data(ai_key, uid, 60)
                    return response.json(0)
                TIME_LOGGER.info(f'uid={uid},sagemakerAI识别失败{ai_result}')

            push_thread = threading.Thread(target=self.image_label_detection,
                                           kwargs={'ai_server': ai_server, 'uid': uid, 'file_list': file_list,
                                                   'detect_group': detect_group, 'n_time': n_time,
                                                   'uid_push_qs': uid_push_qs,
                                                   'channel': channel})
            push_thread.start()  # AI识别异步存表&推送
            redis_obj.set_data(ai_key, uid, 60)
            return response.json(0)
        except Exception as e:
            print(e)
            data = {
                'errLine': e.__traceback__.tb_lineno,
                'errMsg': repr(e)
            }
            TIME_LOGGER.info(f'rekognition识别errMsg={data}')
            return response.json(48, data)

    def async_message_push(self, sage_maker, uid, n_time, uid_push_qs, channel, res, file_list, notify):
        # 保存推送消息
        sage_maker.save_push_message(uid, n_time, uid_push_qs, channel, res, file_list, notify)

    def add_push_cache(self, key, redis_obj, cache_push_data, push_interval):
        """
        推送间隔缓存设置
        """
        if push_interval > 0:
            if cache_push_data:  # 缓存存在
                interval = json.loads(cache_push_data)['interval']
                if interval != push_interval:
                    push_data = {'interval': push_interval}
                    redis_obj.set_data(key=key, val=json.dumps(push_data), expire=push_interval)
            else:  # 缓存不存在
                push_data = {'interval': push_interval}
                redis_obj.set_data(key=key, val=json.dumps(push_data), expire=push_interval)

    def image_label_detection(self, ai_server, uid, file_list, detect_group,
                              n_time, uid_push_qs, channel):
        """
        :param ai_server: AI服务类型
        :param uid: 用户uid
        :param file_list: 图片base64列表
        :param detect_group: 识别组
        :param n_time: 时间戳
        :param uid_push_qs: 推送数据
        :param channel: 推送通道
        :return:
        """
        try:
            start_time = time.time()

            redis_obj = RedisObject(db=6)
            APP_NOTIFY_KEY = f'ASJ:NOTIFY:PUSH:{uid}:{channel}'  # 推送间隔缓存KEY
            push_cache_data = redis_obj.get_data(APP_NOTIFY_KEY)
            is_push = False if push_cache_data else True

            notify_data = uid_push_qs[0]['uid_set__msg_notify']

            # APP推送提醒状态
            notify = self.is_ai_push(uid, notify_data) if is_push else is_push

            TIME_LOGGER.info(f'*****现执行Reko,uid={uid}识别类型={ai_server}')
            dir_path = os.path.join(BASE_DIR, 'static/ai/' + uid + '/' + str(n_time))
            if not os.path.exists(dir_path):
                os.makedirs(dir_path)
            file_path_list = []
            for i, val in enumerate(file_list):
                val = val.replace(' ', '+')
                val = base64.b64decode(val)
                file_path = "{dir_path}/{n_time}_{i}.jpg".format(dir_path=dir_path, n_time=n_time, i=i)
                file_path_list.append(file_path)
                with open(file_path, 'wb') as f:
                    f.write(val)
                    f.close()
            image_size = 0  # 每张小图片的大小，等于0是按原图大小进行合并
            image_colnum = 1  # 合并成一张图后，一行有几个小图
            image_size = MergePic.merge_images(dir_path, image_size, image_colnum)
            photo = open(dir_path + '.jpg', 'rb')  # 打开合成图

            # rekognition识别合成图片
            maxLabels = 50  # 最大标签
            minConfidence = 80  # 置信度

            client = boto3.client(
                'rekognition',
                aws_access_key_id=AWS_ACCESS_KEY_ID[1],
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY[1],
                region_name='us-east-1')
            # 执行AWS Rekognition:
            rekognition_res = client.detect_labels(
                Image={'Bytes': photo.read()},
                MaxLabels=maxLabels,
                MinConfidence=minConfidence)
            photo.close()
            if rekognition_res['ResponseMetadata']['HTTPStatusCode'] != 200:
                return False
            end_time = time.time()

            labels = self.labelsCoords(detect_group, rekognition_res, image_size)  # 检查标签是否符合用户选择的识别类型
            TIME_LOGGER.info(f'uid={uid},{(end_time - start_time)}s,rekognition Result={labels}')

            # 将识别结果存到S3以及DynamoDB
            # AiView.store_image_results_to_dynamo_and_s3(file_path_list, uid, channel, n_time, labels, rekognition_res)
            eventType = labels['eventType']
            label_str = ','.join(labels['label_list'])
            new_bounding_box_dict = labels['new_bounding_box_dict']

            # 上传缩略图到s3
            file_dict = {}
            for i, val in enumerate(file_path_list):
                file_dict[val] = "{uid}/{channel}/{n_time}_{i}.jpeg".format(uid=uid, channel=channel,  # 封面图
                                                                            n_time=n_time, i=i)
            self.upload_s3(file_dict, dir_path)

            # 设置推送间隔缓存
            self.add_push_cache(APP_NOTIFY_KEY, redis_obj, push_cache_data,
                                uid_push_qs[0]['uid_set__new_detect_interval'])

            self.save_message_and_push(eventType, uid, n_time, uid_push_qs, channel,
                                       label_str, new_bounding_box_dict, notify)

            AiView.save_cloud_ai_tag(uid, int(n_time), eventType, 0)
        except Exception as e:
            data = {
                'errLine': e.__traceback__.tb_lineno,
                'errMsg': repr(e)
            }
            TIME_LOGGER.info(f'rekognition识别errMsg={data}')

    def save_message_and_push(self, eventType, uid, n_time, uid_push_qs, channel, label_str, new_bounding_box_dict,
                              notify):
        """
        保存消息以及推送
        """
        uid_push_list = []
        for qs in uid_push_qs:
            uid_push_list.append(qs)

        nickname = uid_push_list[0]['uid_set__nickname']
        if not nickname:
            nickname = uid

        userID_ids = []
        region = 4 if CONFIG_INFO == CONFIG_EUR else 3
        for up in uid_push_list:
            push_type = up['push_type']
            appBundleId = up['appBundleId']
            token_val = up['token_val']
            lang = up['lang']
            tz = up['tz']
            if tz is None or tz == '':
                tz = 0
            local_date_time = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang='cn')
            TIME_LOGGER.info('*****AI消息存库{},{},{}'.format(uid, local_date_time, tz))
            # 以下是存库
            userID_id = up["userID_id"]
            if userID_id not in userID_ids:
                now_time = int(time.time())
                EquipmentInfoService.randoms_insert_equipment_info(
                    device_user_id=userID_id,
                    event_time=n_time,
                    event_type=eventType,
                    device_uid=uid,
                    device_nick_name=nickname,
                    channel=channel,
                    alarm=label_str,
                    is_st=3,
                    add_time=now_time,
                    storage_location=region,
                    border_coords=json.dumps(new_bounding_box_dict)
                )
                userID_ids.append(userID_id)

            if not notify:  # 不推送
                continue
            # 推送标题
            msg_title = self.get_msg_title(appBundleId=appBundleId, nickname=nickname)
            # 推送内容
            msg_text = self.get_msg_text(channel=channel, n_time=n_time, lang=lang, tz=tz, label_list=label_str)
            kwargs = {
                'uid': uid,
                'channel': channel,
                'event_type': eventType,
                'n_time': n_time,
                'appBundleId': appBundleId,
                'token_val': token_val,
                'msg_title': msg_title,
                'msg_text': msg_text,
            }
            try:
                # 推送消息
                if push_type == 0:  # ios apns
                    self.do_apns(**kwargs)
                elif push_type == 1:  # android gcm
                    self.do_fcm(**kwargs)
                elif push_type == 2:  # android jpush
                    self.do_jpush(**kwargs)
            except Exception as e:
                TIME_LOGGER.info('*****error,uid={uid},errLine={errLine}, errMsg={errMsg}'
                                 .format(uid=uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
                continue

    def is_ai_push(self, uid, app_push_config):
        """
        是否进行APP消息提醒
        @return: True|False
        """
        try:
            if not app_push_config:
                return True

            is_push = app_push_config['appPush']
            if is_push != 1:  # 1：进行APP提醒，其它则不执行APP提醒
                return False

            all_day = app_push_config['pushTime']['allDay']

            if all_day == 0:  # 1：全天提醒，0：自定义时间提醒
                push_time_config = app_push_config['pushTime']
                # 计算当前时间是否在自定义消息提醒范围内
                if not DevicePushService.is_push_notify_allowed_now(push_time_config):
                    return False
            # 在开启接收APP消息提醒时，判断是否勾选云端AI消息提醒
            return app_push_config['eventTypes']['aiCloud'] == 1
        except Exception as e:
            TIME_LOGGER.info('*****error,uid={uid},errLine={errLine}, errMsg={errMsg}'
                             .format(uid=uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
            return True

    def del_path(self, path):
        try:
            if not os.path.exists(path):
                return
            if os.path.isfile(path):
                os.remove(path)
            else:
                items = os.listdir(path)
                for f in items:
                    c_path = os.path.join(path, f)
                    if os.path.isdir(c_path):
                        self.del_path(c_path)
                    else:
                        os.remove(c_path)
                os.rmdir(path)
        except Exception as e:
            print(repr(e))

    ## 检查是否有符合条件的标签,并且返回标签坐标位置信息
    def labelsCoords(self, user_detect_group, rekognition_res, image_size):
        logger = logging.getLogger('info')
        labels = rekognition_res['Labels']
        label_name = []
        label_list = []
        logger.info('--------识别到的标签-------')
        logger.info(labels)
        all_labels_type = {
            '1': ['Person', 'Human'],  # 人
            '2': ['Pet', 'Dog', 'Canine', 'Animal', 'Puppy', 'Cat'],  # 动物
            '3': ['Vehicle', 'Car', 'Transportation', 'Automobile', 'Bus'],  # 车
            '4': ['Package', 'Carton', 'Cardboard', 'Package Delivery']  # 包裹
        }
        # 找出识别的所有标签
        for label in labels:
            label_name.append(label['Name'])
            for Parents in label['Parents']:
                label_name.append(Parents['Name'])

        logger.info('标签名------')
        logger.info(label_name)
        # 删除用户没有选择的ai识别类型, 并且得出最终识别结果
        user_detect_list = user_detect_group.split(',')
        user_detect_list = [i.strip() for i in user_detect_list]
        conform_label_list = []
        conform_user_d_group = set()
        for key, label_type_val in all_labels_type.items():
            if key in user_detect_list:
                for label in label_type_val:
                    if label in label_name:
                        conform_user_d_group.add(key)
                        conform_label_list.append(label)
        # 找出标签边框线位置信息
        boundingBoxList = []
        for label in labels:
            if label['Name'] in conform_label_list:
                for boundingBox in label['Instances']:
                    boundingBoxList.append(boundingBox['BoundingBox'])

        # 找出边框位置信息对应的单图位置并重新计算位置比
        merge_image_height = image_size['height']
        # merge_image_width = image_size['width']
        single_height = merge_image_height // image_size['num']
        new_bounding_box_dict = {}
        new_bounding_box_dict['file_0'] = []
        new_bounding_box_dict['file_1'] = []
        new_bounding_box_dict['file_2'] = []
        # new_bounding_box_dict['file_3'] = []
        for k, val in enumerate(boundingBoxList):
            boundingBoxTop = merge_image_height * val['Top']
            # 找出当前边框属于哪张图片范围
            boxDict = {}
            for i in range(image_size['num']):
                min = i * single_height  # 第n张图
                max = (i + 1) * single_height
                if boundingBoxTop >= min and boundingBoxTop <= max:
                    # print("属于第{i}张图".format(i=i+1))
                    boxDict['Width'] = val['Width']
                    boxDict['Height'] = merge_image_height * val['Height'] / single_height
                    boxDict['Top'] = ((merge_image_height * val['Top']) - (
                            i * single_height)) / single_height  # 减去前i张图片的高度
                    boxDict['Left'] = val['Left']
                    new_bounding_box_dict["file_{i}".format(i=i)].append(boxDict)
        # exit(new_bounding_box_list)
        conform_user_d_group = list(conform_user_d_group)
        if len(conform_user_d_group) > 0:
            conform_user_d_group.sort()
            # 集成识别标签
            for label_key in conform_user_d_group:
                label_list.append(AI_IDENTIFICATION_TAGS_DICT[label_key])
            eventType = ''.join(conform_user_d_group)  # 组合类型
        else:
            eventType = ''

        logger.info('------conform_user_d_group------ {}'.format(conform_user_d_group))
        logger.info('------label_list------ {}'.format(label_list))

        return {'eventType': eventType, 'label_list': label_list,
                'new_bounding_box_dict': new_bounding_box_dict}

    def upload_s3(self, file_dict, dir_path):
        try:
            # if CONFIG_INFO == CONFIG_US or CONFIG_INFO == CONFIG_EUR:
            #     # 存国外
            #     aws_key = AWS_ACCESS_KEY_ID[1]
            #     aws_secret = AWS_SECRET_ACCESS_KEY[1]
            #     session = Session(aws_access_key_id=aws_key,
            #                       aws_secret_access_key=aws_secret,
            #                       region_name="us-east-1")
            #     s3 = session.resource("s3")
            #     bucket = "foreignpush"
            # else:
            #     # 存国内
            #     aws_key = AWS_ACCESS_KEY_ID[0]
            #     aws_secret = AWS_SECRET_ACCESS_KEY[0]
            #     session = Session(aws_access_key_id=aws_key,
            #                       aws_secret_access_key=aws_secret,
            #                       region_name="cn-northwest-1")
            #     s3 = session.resource("s3")
            #     bucket = "push"
            #
            # for file_path, upload_path in file_dict.items():
            #     print('-------')
            #     print(file_path)
            #     print('-------')
            #     upload_data = open(file_path, "rb")
            #     # upload_key = "test"
            #     s3.Bucket(bucket).put_object(Key=upload_path, Body=upload_data)
            region = 'eur' if CONFIG_INFO == CONFIG_EUR else 'us'
            oci = OCIObjectStorage(region)
            for file_path, upload_path in file_dict.items():
                upload_data = open(file_path, "rb")
                # OCI上传对象
                oci.put_object(PUSH_BUCKET, upload_path, upload_data, 'image/jpeg')
            return True
        except Exception as e:
            TIME_LOGGER.error('rekoAI上传对象异常errLine={errLine}, errMsg={errMsg}'
                              .format(errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
            return False

    def get_msg_title(self, appBundleId, nickname):
        package_title_config = {
            'com.ansjer.customizedd_a': 'DVS',
            'com.ansjer.zccloud_a': 'ZosiSmart',
            'com.ansjer.zccloud_ab': '周视',
            'com.ansjer.adcloud_a': 'ADCloud',
            'com.ansjer.adcloud_ab': 'ADCloud',
            'com.ansjer.accloud_a': 'ACCloud',
            'com.ansjer.loocamccloud_a': 'Loocam',
            'com.ansjer.loocamdcloud_a': 'Anlapus',
            'com.ansjer.customizedb_a': 'COCOONHD',
            'com.ansjer.customizeda_a': 'Guardian365',
            'com.ansjer.customizedc_a': 'PatrolSecure',
        }
        if appBundleId in package_title_config.keys():
            return package_title_config[appBundleId] + '(' + nickname + ')'
        else:
            return nickname

    def get_msg_text(self, channel, n_time, lang, tz, label_list):
        n_date = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang=lang)
        if lang == 'cn':
            msg = '摄像头AI识别到了{}'.format(label_list)
            send_text = '{msg} 通道:{channel} 日期:{date}'.format(msg=msg, channel=channel, date=n_date)
        else:
            msg = 'Camera AI recognizes {}'.format(label_list)
            send_text = '{msg} channel:{channel} date:{date}'.format(msg=msg, channel=channel, date=n_date)
        return send_text

    @classmethod
    def save_cloud_ai_tag(cls, uid, event_time, types, week=0):
        """
        保存云存AI标签
        """
        try:
            types = str(types)
            if not types:
                return False
            n_time = int(time.time())
            vod_hls_tag = {"uid": uid, "ai_event_time": event_time, "created_time": n_time, 'tab_num': int(week)}
            vod_tag_vo = VodHlsTag.objects.create(**vod_hls_tag)
            tag_list = []
            if len(types) > 1:
                for i in range(1, len(types) + 1):
                    ai_type = MessageTypeEnum(int(types[i - 1:i]))
                    vod_tag_type_vo = VodHlsTagType(tag_id=vod_tag_vo.id, created_time=n_time, type=ai_type.value)
                    tag_list.append(vod_tag_type_vo)
            else:
                ai_type = MessageTypeEnum(int(types))
                vod_tag_type_vo = {"tag_id": vod_tag_vo.id, "created_time": n_time, "type": ai_type.value}
                VodHlsTagType.objects.create(**vod_tag_type_vo)
            if tag_list:
                VodHlsTagType.objects.bulk_create(tag_list)
            return True
        except Exception as e:
            print('AI标签存储异常详情,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
            return False

    def do_jpush(self, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text):
        app_key = JPUSH_CONFIG[appBundleId]['Key']
        master_secret = JPUSH_CONFIG[appBundleId]['Secret']
        # 此处换成各自的app_key和master_secre
        _jpush = jpush.JPush(app_key, master_secret)
        push = _jpush.create_push()
        push.audience = jpush.registration_id(token_val)
        push_data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "",
                     "received_at": n_time, "sound": "sound.aif", "uid": uid, "zpush": "1", "channel": channel}
        android = jpush.android(alert=msg_text, priority=1, style=1, alert_type=7,
                                big_text=msg_text, title=msg_title,
                                extras=push_data)
        push.notification = jpush.notification(android=android)
        push.platform = jpush.all_
        res = push.send()
        print(res)
        return res.status_code

    def do_fcm(self, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text):
        try:

            serverKey = FCM_CONFIG[appBundleId]
            push_service = FCMNotification(api_key=serverKey)
            data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "",
                    "received_at": n_time, "sound": "sound.aif", "uid": uid, "zpush": "1", "channel": channel}
            result = push_service.notify_single_device(registration_id=token_val, message_title=msg_title,
                                                       message_body=msg_text, data_message=data,
                                                       extra_kwargs={
                                                           'default_vibrate_timings': True,
                                                           'default_sound': True,
                                                           'default_light_settings': True
                                                       })
            print('fcm push ing')
            print(result)
            return result
        except Exception as e:
            return 'serverKey abnormal'

    def do_apns(self, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text):
        logger = logging.getLogger('info')
        logger.info("进来do_apns函数了")
        logger.info(token_val)
        logger.info(APNS_MODE)
        logger.info(os.path.join(BASE_DIR, APNS_CONFIG[appBundleId]['pem_path']))
        try:
            cli = apns2.APNSClient(mode=APNS_MODE,
                                   client_cert=os.path.join(BASE_DIR, APNS_CONFIG[appBundleId]['pem_path']))
            push_data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "",
                         "received_at": n_time, "sound": "", "uid": uid, "zpush": "1", "channel": channel}
            alert = apns2.PayloadAlert(body=msg_text, title=msg_title)
            payload = apns2.Payload(alert=alert, custom=push_data, sound="default")
            n = apns2.Notification(payload=payload, priority=apns2.PRIORITY_LOW)
            res = cli.push(n=n, device_token=token_val, topic=appBundleId)

            if res.status_code == 200:
                return res.status_code
            else:
                logger.info('apns push fail')
                logger.info(res.reason)
                return res.status_code
        except (ValueError, ArithmeticError):
            return 'The program has a numeric format exception, one of the arithmetic exceptions'
        except Exception as e:
            print(repr(e))
            logger.info(repr(e))
            return repr(e)

    @staticmethod
    def store_image_results_to_dynamo_and_s3(file_path_list, uid, channel, n_time, labels_data,
                                             reko_result):
        """
        将图片识别结果存储到dynamoDB并且存储到S3
        @param file_path_list: 图片名称列表
        @param uid: 设备uid
        @param channel: 设备通道号
        @param n_time: 设备触发移动侦测时间戳
        @param labels_data: 标签数据（经过reko_result结果进行计算后的数据）
        @param reko_result: rekognition 响应结果
        @return: 保存结果
        """
        logger = logging.getLogger('info')
        try:
            file_dict = {}
            for i, val in enumerate(file_path_list):
                file_dict[val] = "{uid}/{channel}/{n_time}_{i}.jpeg".format(uid=uid, channel=channel,  # 封面图
                                                                            n_time=n_time, i=i)
            if not reko_result:
                logger.info('{}识别结果为空'.format(uid))
                return False
            if CONFIG_INFO != CONFIG_US:  # 目前只上美洲
                return False
            # 存美洲
            session = Session(aws_access_key_id=AWS_ACCESS_KEY_ID[1],
                              aws_secret_access_key=AWS_SECRET_ACCESS_KEY[1],
                              region_name="us-west-1")
            s3 = session.resource("s3")
            bucket = "rekognition-pic-results"
            # 上传到S3 rekognition-pic-results
            for file_path, upload_path in file_dict.items():
                logger.info('{}文件路径{}'.format(uid, file_path))
                upload_data = open(file_path, "rb")
                s3.Bucket(bucket).put_object(Key=upload_path, Body=upload_data)
            # reko结果存储到dynamoDB
            event_type = 0
            new_bounding_box_dict = ''
            if len(labels_data['label_list']) > 0:
                event_type = int(labels_data['eventType'])
                new_bounding_box_dict = json.dumps(labels_data['new_bounding_box_dict'])

            table_name = 'asj_push_message'  # 表名称
            dynamo = DynamodbObject(AWS_ACCESS_KEY_ID[1], AWS_SECRET_ACCESS_KEY[1], 'us-west-1')
            item = {'device_uid': {'S': uid},  # 设备uid
                    'event_time': {'N': str(n_time)},  # 设备触发时间戳，也用作S3资源对象名前缀
                    'ai_coordinate': {'S': new_bounding_box_dict},  # ai坐标框信息
                    'channel': {'N': str(channel)},  # 设备通道号
                    'event_type': {'N': str(event_type)},  # 事件类型
                    'is_pic': {'N': '3'},  # 1:图片,2:视频,3:多图
                    'reko_result': {'S': json.dumps(reko_result)},  # reko识别结果
                    'storage_region': {'N': '2'},  # 存储平台1:阿里云,2:AWS
                    'create_time': {'N': str(int(time.time()))}  # 记录创建时间
                    }
            result = dynamo.put_item(table_name, item)
            logger.info('{}识别后存S3与DynamoDB成功{}'.format(uid, result))
            return True
        except Exception as e:
            logger.info('{}识别后存S3与DynamoDB失败：{}'.format(uid, repr(e)))
            return False

    @staticmethod
    def async_detection_image_label(sage_maker, uid, n_time, uid_push_qs,
                                    channel, file_list, notify, detect_group):
        images = file_list
        final_results = AiView.get_nova_tag_recognition(file_list, uid)
        if not final_results:
            return
        res = sage_maker.get_table_name(uid, final_results, detect_group)
        for i in range(len(images)):
            images[i] = images[i].replace(' ', '+')
        sage_maker.save_push_message(uid, n_time, uid_push_qs, channel, res, images, notify)



    @staticmethod
    def get_nova_tag_recognition(base64_images, uid):
        TIME_LOGGER.info(f'uid:{uid} Nova标签识别开始')
        try:
            redis_obj = RedisObject(db=6)
            now = datetime.now()
            year_month_day = now.strftime("%Y%m%d")  # 如20241005（单日标识）
            daily_total_key = f"api:recognition:nova:daily:{year_month_day}:total"
            # 使用incr命令保证原子操作
            redis_obj.incr(daily_total_key, 1, 3600 * 24 * 7)
        except Exception as e:
            # Redis计数失败不影响主业务（降级处理），仅打日志
            TIME_LOGGER.error(f"{uid}Redis统计接口访问量失败：{repr(e)}")
        # 定义要检测的类别
        categories_to_detect = ["person", "car", "pet", "package"]
        # --- 记录process_image_batch执行时间 ---
        start_time = time.perf_counter()  # 高精度计时
        final_results = nova_image_tag_instance.process_image_batch(
            base64_images,
            categories_to_detect,
            uid
        )
        # 无论是否发生异常，都记录执行时间
        end_time = time.perf_counter()
        execution_time = end_time - start_time
        TIME_LOGGER.info(
            f'uid:{uid} Nova标签识别批量处理完成 | '
            f'执行时间: {execution_time:.4f}秒'
        )
        return final_results

    @staticmethod
    def ai_push_count(uid, date_str):
        """ 记录 AI 调用次数（按月累计） """
        try:
            redis_obj = RedisObject(db=7)
            month_str = date_str[:6]  # "202511"
            key = f"ai:push:monthly:{month_str}"
            ttl = 3600 * 24 * 90
            redis_obj.hash_field_increment(key, uid, 1, ttl)
        except Exception as e:
            TIME_LOGGER.error(f"AI调用计数失败 uid={uid} error={repr(e)}")

    @staticmethod
    def ai_push_stats(request_dict, response):
        uid = request_dict.get('uid', None)
        start = request_dict.get('start_time')
        end = request_dict.get('end_time')

        if not all([start, end]):
            return response.json(444)

        try:
            start_dt = datetime.strptime(start, "%Y-%m")
            end_dt = datetime.strptime(end, "%Y-%m")
        except ValueError:
            return response.json(444, '月份格式错误，必须为 YYYY-MM')

        if start_dt > end_dt:
            return response.json(444, '开始月份不能晚于结束月份')

        try:
            redis_obj = RedisObject(db=7)
            total = 0
            monthly = []

            # 当前月份从开始月份的1号开始
            cur = start_dt
            while cur <= end_dt:
                month_str = cur.strftime("%Y%m")
                key = f"ai:push:monthly:{month_str}"
                month_data = {
                    "month": cur.strftime("%Y-%m"),
                    "total_count": 0,  # 当月总次数
                }

                if uid:
                    #只返回该设备的数据
                    count = redis_obj.CONN.hget(key, uid)
                    count = int(count) if count else 0
                    month_data["total_count"] = count
                else:
                    #返回所有设备的明细和总次数
                    data = redis_obj.CONN.hgetall(key)
                    user_list = []
                    total_month = 0
                    for user_uid, cnt in data.items():
                        uid_str = user_uid.decode('utf-8')
                        count_int = int(cnt.decode('utf-8'))
                        user_list.append({"uid": uid_str, "count": count_int})
                        total_month += count_int
                    month_data["total_count"] = total_month
                    month_data["users"] = user_list

                total += month_data["total_count"]
                monthly.append(month_data)

                cur += timedelta(days=32)
                cur = cur.replace(day=1)

            result = {
                "uid": uid,
                "total": total,  # 所有月份的总和
                "monthly": monthly  # 每个月的明细
            }
            return response.json(0, result)
        except Exception as e:
            return response.json(500, f"error_line:{e.__traceback__.tb_lineno}, error_msg:{repr(e)}")