import time import boto3 import botocore import oss2 from botocore import client from django.core.paginator import Paginator from django.db.models import Q from django.views.generic.base import View from obs import ObsClient from Ansjer.config import CONFIG_EUR, CONFIG_US, HUAWEICLOUD_PUSH_BUKET, HUAWEICLOUD_AK, HUAWEICLOUD_SK, \ HUAWEICLOUD_OBS_SERVER from Ansjer.config import OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, \ SERVER_TYPE from Controller.DetectControllerV2 import DetectControllerViewV2 from Model import models from Model.models import Device_Info, VodBucketModel, DeviceTypeModel from Object.OCIObjectStorage import OCIObjectStorage from Object.RedisObject import RedisObject from Object.ResponseObject import ResponseObject from Object.TokenObject import TokenObject from Object.utils import LocalDateTimeUtil from Service.EquipmentInfoService import EquipmentInfoService from Service.VodHlsService import SplitVodHlsObject class MassageView(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request, request.GET, operation) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request, request.POST, operation) def validation(self, request, request_dict, operation): language = request_dict.get('language', 'en') response = ResponseObject(language, 'pc') if operation == 'XXX': return 0 else: tko = TokenObject(request.META.get('HTTP_AUTHORIZATION'), returntpye='pc') if tko.code != 0: return response.json(tko.code) if operation == 'queryInfoList': return self.query_info_list(request_dict, response) else: return response.json(414) def query_info_list(self, request_dict, response): """ 查询推送数据 @param request_dict: 请求参数 @request_dict uids: 设备id @request_dict eventType: 事件类型 @request_dict eventType: userID @request_dict startTime: 开始时间戳 @request_dict endTime: 结束时间戳 @request_dict pageNo: 开始时间戳 @request_dict pageSize: 结束时间戳 @param response: @return: """ try: uids = request_dict.get('uids', None) event_type = request_dict.get('eventType', None) user_id = request_dict.get('userID', None) # 时间 start_time = request_dict.get('startTime', None) end_time = request_dict.get('endTime', None) # 分页 page = int(request_dict.get('pageNo', None)) size = int(request_dict.get('pageSize', None)) # 区域 if SERVER_TYPE == 'Ansjer.cn_config.formal_settings' or SERVER_TYPE == 'Ansjer.cn_config.test_settings': region = 2 else: region = 1 query = Q() if not start_time and not end_time: # 默认查询近七天内数据 end_time = int(time.time()) start_time = LocalDateTimeUtil.get_before_days_timestamp(end_time, 7) query &= Q(event_time__range=(start_time, end_time)) elif start_time and end_time: query &= Q(event_time__range=(start_time, end_time)) else: response.json(10, "需要给出一个时间段") if user_id is None and uids is None: return response.json(0, {"list": [], "total": 0}) # 过滤条件 if user_id is not None and user_id != "": query &= Q(device_user_id=user_id) if uids is not None and uids != "": uid_list = uids.split(',') query &= Q(device_uid__in=uid_list) if event_type is not None: event_type_list = EquipmentInfoService.get_comb_event_type(event_type) event_type_list = list(set(event_type_list)) tags = EquipmentInfoService.get_event_tag(event_type) if event_type_list: query &= Q(event_type__in=event_type_list) tags = '' query &= Q(event_tag__regex=tags) elif tags: query &= Q(event_tag__regex=tags) # 联表查询 querysets = [] for i in range(1, 41): table_name = f'EquipmentInfo{i}' model_class = getattr(models, table_name) annotated_queryset = model_class.objects.filter(query) querysets.append(annotated_queryset) equipment_info_combined_qs = querysets[0].union(*querysets[1:], all=True) # 创建分页对象 equipment_info_combined_qs = equipment_info_combined_qs.order_by("-event_time") paginator = Paginator(equipment_info_combined_qs, size) # 获取请求页的数据 packages_page = paginator.page(page) # 连接存储桶 auth = oss2.Auth(OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET) oss_img_bucket = oss2.Bucket(auth, 'oss-cn-shenzhen.aliyuncs.com', 'apg') # 华为云 obs_client = ObsClient( access_key_id=HUAWEICLOUD_AK, secret_access_key=HUAWEICLOUD_SK, server=HUAWEICLOUD_OBS_SERVER) aws_s3 = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID[1], aws_secret_access_key=AWS_SECRET_ACCESS_KEY[1], config=botocore.client.Config(signature_version='s3v4'), region_name='us-east-1' ) aws_s3_cn = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID[0], aws_secret_access_key=AWS_SECRET_ACCESS_KEY[0], config=botocore.client.Config(signature_version='s3v4'), region_name='cn-northwest-1' ) # 国内生产环境默认不实例OCI对象 oci_eur = OCIObjectStorage(CONFIG_EUR) oci_us = OCIObjectStorage(CONFIG_US) redis_obj = RedisObject(3) # ai消息标识所有组合标签 ai_all_event_type = EquipmentInfoService.get_all_comb_event_type() # 遍历调整返回数据 equipment_info_list = [] for equipment_info in packages_page: uid = equipment_info.device_uid channel = equipment_info.channel event_time = equipment_info.event_time storage_location = equipment_info.storage_location border_coords = equipment_info.border_coords event_tag = equipment_info.event_tag img_url = "" if equipment_info.is_st == 1: thumbspng = '{}/{}/{}.jpeg'.format(uid, channel, event_time) if storage_location == 1: # 阿里云oss img_url = oss_img_bucket.sign_url('GET', thumbspng, 300) elif storage_location == 5: # 华为云 create_res = obs_client.createSignedUrl( method='GET', bucketName=HUAWEICLOUD_PUSH_BUKET, objectKey=thumbspng, expires=300) img_url = create_res.signedUrl elif storage_location in [3, 4]: prefix_name = f'{uid}/' oci = oci_eur if storage_location == 4 else oci_us img_url = DetectControllerViewV2.oci_object_url(oci, redis_obj, uid, prefix_name) if img_url: img_url = img_url + thumbspng else: params = {'Key': thumbspng} if region == 1: # AWS国外 params['Bucket'] = 'foreignpush' img_url = aws_s3.generate_presigned_url( 'get_object', Params=params, ExpiresIn=300) else: # AWS国内 params['Bucket'] = 'push' img_url = aws_s3_cn.generate_presigned_url( 'get_object', Params=params, ExpiresIn=300) img_url = img_url img_list = [img_url] elif equipment_info.is_st == 2: # 列表装载回放时间戳标记 split_vod_hls_obj = SplitVodHlsObject() vodqs = split_vod_hls_obj.get_vod_hls_data( uid=uid, channel=channel, start_time=int(event_time)).values('bucket_id') if not vodqs.exists(): return response.json(173) vod_bucket_qs = VodBucketModel.objects.filter(id=vodqs[0]['bucket_id']).values('bucket', 'endpoint') if not vod_bucket_qs.exists(): return response.json(173) bucket_name = vod_bucket_qs[0]['bucket'] endpoint = vod_bucket_qs[0]['endpoint'] bucket = oss2.Bucket(auth, endpoint, bucket_name) ts = '{}/vod{}/{}/ts0.ts'.format(uid, channel, event_time) if storage_location == 1: # 阿里云oss thumb0 = bucket.sign_url('GET', ts, 3600, params={'x-oss-process': 'video/snapshot,t_0000,w_700'}) thumb1 = bucket.sign_url('GET', ts, 3600, params={'x-oss-process': 'video/snapshot,t_1000,w_700'}) thumb2 = bucket.sign_url('GET', ts, 3600, params={'x-oss-process': 'video/snapshot,t_2000,w_700'}) img_url = "" img_list = [thumb0, thumb1, thumb2] else: params = {'Key': ts} if region == 1: # AWS国外 params['Bucket'] = 'foreignpush' img_url = aws_s3.generate_presigned_url( 'get_object', Params=params, ExpiresIn=300) else: # AWS国内 params['Bucket'] = 'push' img_url = aws_s3_cn.generate_presigned_url( 'get_object', Params=params, ExpiresIn=300) img_list = [img_url] elif equipment_info.is_st == 3 or equipment_info.is_st == 4: # 列表装载回放时间戳标记 img_list = [] for i in range(equipment_info.is_st - 1, -1, -1): thumbspng = '{}/{}/{}_{}.jpeg'.format(uid, channel, event_time, i) if storage_location == 1: # 阿里云oss img_url = oss_img_bucket.sign_url('GET', thumbspng, 300) elif storage_location == 5: # 华为云 create_res = obs_client.createSignedUrl( method='GET', bucketName=HUAWEICLOUD_PUSH_BUKET, objectKey=thumbspng, expires=300) img_url = create_res.signedUrl elif storage_location in [3, 4]: # 国外OCI云 prefix_name = f'{uid}/' oci = oci_eur if storage_location == 4 else oci_us img_url = DetectControllerViewV2.oci_object_url(oci, redis_obj, uid, prefix_name) if img_url: img_url = img_url + thumbspng else: params = {'Key': thumbspng} if region == 1: # 国外AWS params['Bucket'] = 'foreignpush' img_url = aws_s3.generate_presigned_url( 'get_object', Params=params, ExpiresIn=300) else: # 国内AWS params['Bucket'] = 'push' img_url = aws_s3_cn.generate_presigned_url( 'get_object', Params=params, ExpiresIn=300) img_list.append(img_url) else: img_list = [] uid_type = Device_Info.objects.filter(UID=uid).values('Type').first() if uid_type is None: return response.json(10077) device_type = DeviceTypeModel.objects.filter(type=uid_type['Type']).values('name').first() if device_type is None: device_type = {"name": uid_type['Type']} border_coords = '' if border_coords == '' else eval(border_coords) ai_event_type_list = [] # 如果是ai消息类型，则分解eventType, 如：123 -> [1,2,3] if border_coords and event_type in ai_all_event_type: ai_event_type_list = list(map(int, str(event_type))) if EquipmentInfoService.is_combo_tag(event_type, event_tag): ai_event_type_list += EquipmentInfoService.get_combo_types(event_type, event_tag) equipment_info_data = { "uid": uid, "status": equipment_info.status, "answerStatus": equipment_info.answer_status, "alarm": equipment_info.alarm, "isSt": equipment_info.is_st, "storageLocation": storage_location, "devNickName": equipment_info.device_nick_name, "channel": channel, "eventType": equipment_info.event_type, "eventTime": int(event_time), "addTime": equipment_info.add_time, "borderCoords": border_coords, "eventTag": event_tag, "img": img_url, "imgList": img_list, "uidType": device_type["name"], "aiEventTypeList": str(ai_event_type_list), } equipment_info_list.append(equipment_info_data) data = { "list": equipment_info_list, "total": paginator.count } return response.json(0, data) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))