#!/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 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 AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, APNS_MODE, APNS_CONFIG, BASE_DIR, \ JPUSH_CONFIG, FCM_CONFIG from AnsjerPush.config import CONFIG_INFO from Model.models import UidPushModel, AiService, VodHlsTag, VodHlsTagType from Object import MergePic from Object.DynamodbObject import DynamodbObject from Object.ETkObject import ETkObject 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.EquipmentInfoService import EquipmentInfoService # 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) 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()) logger = logging.getLogger('info') logger.info('*****进入into----ai--api,etk={etk}'.format(etk=etk)) if not etk: return response.json(444) try: # 解密uid及判断长度 eto = ETkObject(etk) uid = eto.uid logger.info(f'etk解析uid={uid},n_time={n_time}') 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(): 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: return response.json(444, '缺少第{k}张图'.format(k=k + 1)) # 查询推送数据 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__channel') if not uid_push_qs.exists(): 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 * 15): ai_server = 'rekognition' 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: res = sage_maker.get_table_name(uid, ai_result, AiServiceQuery[0]['detect_group']) if not res: return response.json(0) sage_maker.save_push_message(uid, n_time, uid_push_qs, channel, res, file_list) return response.json(0) logger.info(f'*****执行Reko,uid={uid} run {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') # 打开合成图 # 识别合成图片 maxLabels = 50 # 最大标签 minConfidence = 80 # 置信度 client = boto3.client( 'rekognition', aws_access_key_id='AKIA2E67UIMD6JD6TN3J', aws_secret_access_key='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', 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 response.json(173) labels = self.labelsCoords(detect_group, rekognition_res, image_size) # 检查标签是否符合用户选择的识别类型 logger.info('*****执行完Rekognition得到labels') # 将识别结果存到S3以及DynamoDB AiView.store_image_results_to_dynamo_and_s3(file_path_list, uid, channel, n_time, labels, rekognition_res) if len(labels['label_list']) == 0: # 需要删除图片 self.del_path(dir_path) self.del_path(dir_path + '.jpg') logger.info('*****uid={},删除图片成功,路径为：{}'.format(uid, dir_path)) return response.json(10055) eventType = labels['eventType'] label_str = ','.join(labels['label_list']) new_bounding_box_dict = labels['new_bounding_box_dict'] logger.info(eventType) logger.info(label_str) # 上传缩略图到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) # 存储消息以及推送 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 = [] 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') 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=2, border_coords=json.dumps(new_bounding_box_dict) ) userID_ids.append(userID_id) # 推送标题 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: logger.info( '*****error,uid={uid},errLine={errLine}, errMsg={errMsg}' .format(uid=uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e))) continue AiView.save_cloud_ai_tag(uid, int(n_time), eventType, 0) return response.json(0) except Exception as e: print(e) data = { 'errLine': e.__traceback__.tb_lineno, 'errMsg': repr(e) } logger.info(f'rekognition识别errMsg={data}') return response.json(48, data) def del_path(self, path): 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) ## 检查是否有符合条件的标签,并且返回标签坐标位置信息 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) # 需要删除图片 logger = logging.getLogger('info') self.del_path(dir_path) self.del_path(dir_path + '.jpg') logger.info('删除图片成功,路径为：{}'.format(dir_path)) return True except Exception as e: print(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