# -*- coding: utf-8 -*- """ @Author ： peng @Time ： 2023-6-7 18:26:35 @File ：IcloudMeal.py """ from Ansjer.config import LOGGER import time import jwt from django.db.models import Sum, Q from django.http import HttpResponse from django.views import View from Model.models import IcloudUseDetails, IcloudService, VodBucketModel, IcloudStorageRecord, Device_Info, Device_User from Object.AWS.AmazonS3Util import AmazonS3Util from Object.RedisObject import RedisObject from Object.ResponseObject import ResponseObject from Object.TokenObject import TokenObject from Ansjer.config import ACCESS_KEY_ID, SECRET_ACCESS_KEY, REGION_NAME, SERVER_DOMAIN, AWS_ACCESS_KEY_ID, \ AWS_SECRET_ACCESS_KEY, SERVER_DOMAIN_SSL, OAUTH_ACCESS_TOKEN_SECRET from Object.m3u8generate import PlaylistGenerator from Service.VodHlsService import SplitVodHlsObject class IcloudServiceView(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.GET, operation, request) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.POST, operation, request) def validation(self, request_dict, operation, request): response = ResponseObject() if operation == 'icloudPlay': # 云盘播放 return self.icloud_play(request_dict, response) else: tko = TokenObject(request.META.get('HTTP_AUTHORIZATION')) response.lang = tko.lang user_id = tko.userID if tko.code != 0: return response.json(tko.code) if operation == 'vodUpload': # 云存上传云盘 return self.vod_upload(request_dict, response, user_id) elif operation == 'getUploadUrl': # 获取s3上传地址 return self.get_upload_url(request_dict, response, user_id) elif operation == 'localUpload': # 本地上传云盘 return self.local_upload(request_dict, response, user_id) elif operation == 'deleteIcloudRecord': # 删除云盘记录 return self.delete_icloud_record(request_dict, response, user_id) elif operation == 'getIcloudList': # 获取云盘播放列表 return self.get_icloud_list(request_dict, response, user_id) elif operation == 'changeIcloudStatus': # 修改云盘开关状态 return self.do_change_icloud_status(request_dict, response, user_id) else: return response.json(404) @staticmethod def vod_upload(request_dict, response, user_id): """ 云存上传云盘 @param request_dict: 请求参数 @param user_id: 用户id @request_dict uid: 设备uid @request_dict channel: 设备通道 @request_dict start_time: 播放时间戳 @param response: 响应对象 @return: response """ LOGGER.info('开始云存转移到云盘') uid = request_dict.get('uid', None) channel = request_dict.get('channel', None) time_stamp = request_dict.get('time_stamp', None) if not all([uid, channel, time_stamp]): return response.json(444, {'error param': 'uid,channel,start_time'}) device_qs = Device_Info.objects.filter(UID=uid, userID=user_id).values('NickName') if not device_qs.exists(): return response.json(173) # 查询是否开通云盘 use_details_qs = IcloudUseDetails.objects.filter(user_id=user_id).values('id', 'bucket_id', 'use_size') if not use_details_qs.exists(): return response.json(173) use_details_id = use_details_qs[0]['id'] target_bucket_id = use_details_qs[0]['bucket_id'] use_size = float(use_details_qs[0]['use_size']) now_time = int(time.time()) nickname = device_qs[0]['NickName'] # 判断云盘是否还有容量 all_size = IcloudService.objects.filter(Q(use_details_id=use_details_id), Q(use_status=0), Q(end_time__gt=now_time) | Q(end_time=0)).values( 'size').aggregate(total_size=Sum('size'))['total_size'] all_size = all_size * 1024 if all_size else 0 if use_size > all_size: LOGGER.info('{}用户套餐总容量为：{}，已使用容量为：{}'.format(uid, all_size, use_size)) return response.json(910) split_vod_hls_obj = SplitVodHlsObject() vod_hls = split_vod_hls_obj.get_vod_hls_data(uid=uid, channel=channel, start_time=time_stamp).values( 'bucket_id', 'fg', 'sec') if not vod_hls.exists(): LOGGER.info('{}用户查无此云存：{}'.format(uid, time_stamp)) return response.json(173) source_bucket_id = vod_hls[0]['bucket_id'] fg = int(vod_hls[0]['fg']) sec = vod_hls[0]['sec'] target_bucket_qs = VodBucketModel.objects.filter(id=target_bucket_id).values('bucket') if not target_bucket_qs.exists(): return response.json(173) target_bucket_name = target_bucket_qs[0]['bucket'] source_bucket_qs = VodBucketModel.objects.filter(id=source_bucket_id).values('bucket', 'region', 'mold') if not source_bucket_qs.exists(): return response.json(173) bucket_region = source_bucket_qs[0]['region'] source_bucket_name = source_bucket_qs[0]['bucket'] mold = source_bucket_qs[0]["mold"] ts_list = [] ts_size = 0 try: s3_obj = AmazonS3Util( AWS_ACCESS_KEY_ID[mold], AWS_SECRET_ACCESS_KEY[mold], bucket_region ) # 获取视频封面 thumbs_png = '{uid}/vod{channel}/{time}/Thumb.jpeg'.format(uid=uid, channel=channel, time=time_stamp) thumbs_png_size = s3_obj.get_object_size(source_bucket_name, thumbs_png) if thumbs_png_size: ts_size += thumbs_png_size ts_list.append(thumbs_png) # 获取s3对象，并计算总字节 for i in range(15): shift = (i + 1) * 4 duration = (fg >> shift) & 0xf if duration > 0: ts_file = '{uid}/vod{channel}/{time}/ts{i}.ts'.format(uid=uid, channel=channel, time=time_stamp, i=i) ts_list.append(ts_file) ts_size += s3_obj.get_object_size(source_bucket_name, ts_file) # 获取存储对象字节,单位B ts_size = round(ts_size / 1024 / 1024, 2) # 字节转换为MB单位 temp_size = ts_size + use_size if temp_size > all_size: LOGGER.info('{}用户无空间上传，套餐容量为：{}，使用容量为：{}'.format(uid, all_size, temp_size)) return response.json(910) time_stamp = int(time_stamp) * 1000 # 转换单位为毫秒 icloud_record_qs = IcloudStorageRecord.objects.filter(user_id=user_id, uid=uid, channel=channel, time_stamp=time_stamp, size=ts_size, file_type=1) if not icloud_record_qs.exists(): # 转移云盘，并记录上传记录，更新使用容量 for source_key in ts_list: ts_name = source_key.split('/')[-1] target_key = '{user_id}/ts_file/{uid}/channel{channel}/{time}/{ts_name}'.format(user_id=user_id, uid=uid, channel=channel, time=time_stamp, ts_name=ts_name) s3_obj.copy_single_obj(source_bucket_name, source_key, target_bucket_name, target_key, StorageClass='STANDARD_IA') IcloudStorageRecord.objects.create(user_id=user_id, uid=uid, channel=channel, time_stamp=time_stamp, nickname=nickname, sec=sec, bucket_id=target_bucket_id, fg=fg, size=ts_size, file_type=1) use_details_qs.update(use_size=temp_size) return response.json(0) else: return response.json(174) except Exception as e: LOGGER.info('云存转移云盘异常:{}'.format(repr(e))) return response.json(500) @staticmethod def local_upload(request_dict, response, user_id): """ 本地上传云盘 @param request_dict: 请求参数 @param user_id: 用户id @request_dict uid: 设备uid @request_dict channel: 设备通道 @request_dict time_stamp: 时间戳 @request_dict size: 文件大小 @request_dict file_type: 文件类型 @request_dict sec: 时长 @param response: 响应对象 @return: response """ LOGGER.info('开始本地转移到云盘') uid = request_dict.get('uid', None) channel = request_dict.get('channel', None) time_stamp = request_dict.get('time_stamp', None) size = request_dict.get('size', None) file_type = request_dict.get('file_type', None) sec = request_dict.get('sec', 0) if not all([uid, channel, time_stamp, size]): return response.json(444, {'error param': 'uid,channel,time_stamp,size'}) device_qs = Device_Info.objects.filter(UID=uid, userID=user_id).values('NickName') if not device_qs.exists(): return response.json(173) # 查询是否开通云盘 use_details_qs = IcloudUseDetails.objects.filter(user_id=user_id).values('id', 'use_size', 'bucket_id') if not use_details_qs.exists(): return response.json(173) size = float(size) # 单位是MB use_details_id = use_details_qs[0]['id'] bucket_id = use_details_qs[0]['bucket_id'] use_size = float(use_details_qs[0]['use_size']) now_time = int(time.time()) nickname = device_qs[0]['NickName'] # 判断云盘是否还有容量 all_size = IcloudService.objects.filter(Q(use_details_id=use_details_id), Q(use_status=0), Q(end_time__gt=now_time) | Q(end_time=0)).values( 'size').aggregate(total_size=Sum('size'))['total_size'] all_size = all_size * 1024 if all_size else 0 if use_size > all_size: LOGGER.info('{}用户套餐总容量为：{}，已使用容量为：{}'.format(uid, all_size, use_size)) return response.json(910) try: new_size = round(use_size + size, 2) icloud_record_qs = IcloudStorageRecord.objects.filter(user_id=user_id, uid=uid, channel=channel, time_stamp=time_stamp, size=size, file_type=file_type) if not icloud_record_qs.exists(): IcloudStorageRecord.objects.create(user_id=user_id, uid=uid, channel=channel, time_stamp=time_stamp, nickname=nickname, sec=sec, bucket_id=bucket_id, size=size, file_type=file_type) use_details_qs.update(use_size=new_size) return response.json(0) else: return response.json(174) except Exception as e: LOGGER.info('本地转移云盘异常:{}'.format(repr(e))) return response.json(500) @staticmethod def get_upload_url(request_dict, response, user_id): """ 获取s3上传地址 @param request_dict: 请求参数 @param user_id: 用户id @request_dict time_stamp: 时间戳 @request_dict file_type: 文件类型 @request_dict uid: 设备uid @request_dict channel: 设备通道 @param response: 响应对象 @return: response """ LOGGER.info('获取s3上传地址') time_stamp = request_dict.get('time_stamp', None) file_type = request_dict.get('file_type', None) uid = request_dict.get('uid', None) channel = request_dict.get('channel', None) if not all([time_stamp, uid, channel]): return response.json(444, {'error param': 'time_stamp,channel,uid'}) device_qs = Device_Info.objects.filter(UID=uid, userID=user_id) if not device_qs.exists(): return response.json(173) # 查询是否开通云盘 use_details_qs = IcloudUseDetails.objects.filter(user_id=user_id).values('id', 'use_size', 'bucket_id') if not use_details_qs.exists(): return response.json(173) use_details_id = use_details_qs[0]['id'] use_size = use_details_qs[0]['use_size'] bucket_id = use_details_qs[0]['bucket_id'] now_time = int(time.time()) # 判断云盘是否还有容量 all_size = IcloudService.objects.filter(Q(use_details_id=use_details_id), Q(use_status=0), Q(end_time__gt=now_time) | Q(end_time=0)).values( 'size').aggregate(total_size=Sum('size'))['total_size'] all_size = all_size * 1024 if all_size else 0 excess_size = all_size - use_size if excess_size < 0: LOGGER.info('{}用户套餐总容量为：{}，已使用容量为：{}'.format(user_id, all_size, use_size)) return response.json(910) bucket_qs = VodBucketModel.objects.filter(id=bucket_id).values('bucket', 'region', 'mold') if not bucket_qs.exists(): return response.json(173) bucket_name = bucket_qs[0]['bucket'] bucket_region = bucket_qs[0]['region'] mold = bucket_qs[0]['mold'] try: s3_obj = AmazonS3Util( AWS_ACCESS_KEY_ID[mold], AWS_SECRET_ACCESS_KEY[mold], bucket_region ) if file_type == '0': # 图片 key_name = '{user_id}/image_file/{uid}/channel{channel}/{time}/{time}.jpeg'.format(user_id=user_id, uid=uid, channel=channel, time=time_stamp) elif file_type == '2': key_name = '{user_id}/mp4_file/{uid}/channel{channel}/{time}/{time}.mp4'.format(user_id=user_id, uid=uid, channel=channel, time=time_stamp) else: return response.json(444, {'error param': 'file_type'}) upload_url = s3_obj.generate_put_obj_url(bucket_name, key_name, 'STANDARD_IA') return response.json(0, {'uploadUrl': upload_url, 'excessSize': excess_size}) except Exception as e: LOGGER.info('获取s3上传地址异常:{}'.format(repr(e))) return response.json(500) @staticmethod def delete_icloud_record(request_dict, response, user_id): """ 删除云盘记录 @param request_dict: 请求参数 @param user_id: 用户id @request_dict time_stamp: 时间戳 @request_dict uid: 设备uid @request_dict channel: 设备通道 @request_dict file_type: 文件类型 @param response: 响应对象 @return: response """ LOGGER.info('删除云盘记录') record_ids = request_dict.get('record_ids', None) if not all([record_ids]): return response.json(444, {'error param': 'record_id'}) record_id_list = record_ids.split(',') # 查询是否开通云盘 use_details_qs = IcloudUseDetails.objects.filter(user_id=user_id).values('use_size', 'bucket_id') if not use_details_qs.exists(): return response.json(173) use_size = float(use_details_qs[0]['use_size']) bucket_id = use_details_qs[0]['bucket_id'] icloud_record_qs = IcloudStorageRecord.objects.filter(user_id=user_id, id__in=record_id_list) if not icloud_record_qs.exists(): return response.json(173) record_qs = icloud_record_qs.values('size', 'uid', 'channel', 'time_stamp', 'file_type') bucket_qs = VodBucketModel.objects.filter(id=bucket_id).values('bucket', 'region', 'mold') if not bucket_qs.exists(): return response.json(173) bucket_name = bucket_qs[0]['bucket'] bucket_region = bucket_qs[0]['region'] mold = bucket_qs[0]['mold'] try: s3_obj = AmazonS3Util( AWS_ACCESS_KEY_ID[mold], AWS_SECRET_ACCESS_KEY[mold], bucket_region ) for item in record_qs: size = float(item['size']) uid = item['uid'] channel = item['channel'] time_stamp = item['time_stamp'] file_type = item['file_type'] use_size = round(use_size - size, 2) if file_type == 0: # 图片 key_name = '{user_id}/image_file/{uid}/channel{channel}/{time}'.format(user_id=user_id, uid=uid, channel=channel, time=time_stamp) elif file_type == 1: # ts文件 key_name = '{user_id}/ts_file/{uid}/channel{channel}/{time}'.format(user_id=user_id, uid=uid, channel=channel, time=time_stamp) elif file_type == 2: # mp4文件 key_name = '{user_id}/mp4_file/{uid}/channel{channel}/{time}'.format(user_id=user_id, uid=uid, channel=channel, time=time_stamp) else: return response.json(444, {'error param': 'file_type'}) object_list = s3_obj.get_object_list(bucket_name, key_name) for obj in object_list: s3_obj.delete_obj(bucket_name, obj['Key']) use_size = use_size if use_size > 0 else 0 icloud_record_qs.delete() use_details_qs.update(use_size=use_size) return response.json(0) except Exception as e: LOGGER.info('删除云盘记录异常:{}'.format(repr(e))) return response.json(500) @staticmethod def get_icloud_list(request_dict, response, user_id): """ 获取云盘播放列表 @param request_dict: 请求参数 @param user_id: 用户id @request_dict time_stamp: 时间戳 @request_dict uid: 设备uid @request_dict channel: 设备通道 @param response: 响应对象 @return: response """ LOGGER.info('删除云盘记录') start_time = request_dict.get('start_time', None) end_time = request_dict.get('end_time', None) uid = request_dict.get('uid', None) file_type = request_dict.get('file_type', None) page = request_dict.get('page', None) line = request_dict.get('line', None) if not all([page, line]): return response.json(444, {'error param': 'page, line'}) page = int(page) line = int(line) now_time = int(time.time()) icloud_list = [] use_details_qs = IcloudUseDetails.objects.filter(user_id=user_id).values('id') if not use_details_qs.exists(): return response.json(911) use_details_id = use_details_qs[0]['id'] all_size = IcloudService.objects.filter(Q(use_details_id=use_details_id), Q(use_status=0), Q(end_time__gt=now_time) | Q(end_time=0)).values( 'size').aggregate(total_size=Sum('size'))['total_size'] all_size = float(all_size * 1024) if all_size else 0 # 转换单位为MB icloud_record_qs = IcloudStorageRecord.objects.filter(user_id=user_id) if start_time and end_time: # 按时间段查询 icloud_record_qs = icloud_record_qs.filter(time_stamp__gte=start_time, time_stamp__lt=end_time) if uid: # 按设备昵称查询 uid_list = uid.split(',') icloud_record_qs = icloud_record_qs.filter(uid__in=uid_list) if file_type: # 按文件类型查询 file_type_list = file_type.split(',') icloud_record_qs = icloud_record_qs.filter(file_type__in=file_type_list) icloud_record_qs = icloud_record_qs.values('bucket_id', 'sec', 'fg', 'nickname', 'uid', 'channel', 'size', 'id', 'time_stamp', 'file_type').order_by('-time_stamp')[ (page - 1) * line:page * line] if not icloud_record_qs.exists(): return response.json(0, {'icloudList': icloud_list, 'notShow': 0}) bucket_id = icloud_record_qs[0]['bucket_id'] bucket_qs = VodBucketModel.objects.filter(id=bucket_id).values('bucket', 'region', 'mold') if not bucket_qs.exists(): return response.json(173) bucket_name = bucket_qs[0]['bucket'] bucket_region = bucket_qs[0]['region'] mold = bucket_qs[0]['mold'] total_size = 0 try: s3_obj = AmazonS3Util( AWS_ACCESS_KEY_ID[mold], AWS_SECRET_ACCESS_KEY[mold], bucket_region ) for item in icloud_record_qs: uid = item['uid'] channel = item['channel'] time_stamp = item['time_stamp'] file_type = item['file_type'] total_size += float(item['size']) uid_token = TokenObject().encryption(data={'uid': uid, 'user_id': user_id}) res_data = { 'is_show': 1 if total_size < all_size else 0, # 只展示剩余容量的数据，超过的部分不展示，1：展示；0：不展示 'time_stamp': time_stamp, 'icloud_id': item['id'], 'file_type': file_type, 'nick_name': item['nickname'], 'uid': uid } if file_type == 0: # 图片,只返回图片地址 thumbs_png = '{user_id}/image_file/{uid}/channel{channel}/{time}/{time}.jpeg'.format( user_id=user_id, uid=uid, channel=channel, time=time_stamp) response_url = s3_obj.generate_file_obj_url(bucket_name, thumbs_png) res_data['thumb'] = response_url icloud_list.append(res_data) elif file_type == 1: # ts文件，返回播放地址 thumbs_png = '{user_id}/ts_file/{uid}/channel{channel}/{time}/Thumb.jpeg'.format( user_id=user_id, uid=uid, channel=channel, time=time_stamp) response_url = s3_obj.generate_file_obj_url(bucket_name, thumbs_png) ts_num = int(item['fg']) & 0xf res_data['thumb'] = response_url icloud_url = '{server_domain}icloud/service/icloudPlay?' \ 'uid={uid}&channel={channel}&time_stamp={time_stamp}&file_type=1&sign=tktktktk'. \ format(server_domain=SERVER_DOMAIN_SSL, uid=uid_token, channel=channel, time_stamp=time_stamp) res_data['ts_num'] = ts_num res_data['icloud_url'] = icloud_url res_data['sec'] = item['sec'] icloud_list.append(res_data) elif file_type == 2: # mp4文件，返回播放地址 mp4_file = '{user_id}/mp4_file/{uid}/channel{channel}/{time}/{time}.mp4'.format( user_id=user_id, uid=uid, channel=channel, time=time_stamp) response_url = s3_obj.generate_file_obj_url(bucket_name, mp4_file) res_data['icloud_url'] = response_url res_data['sec'] = item['sec'] icloud_list.append(res_data) not_show = total_size - all_size if total_size > all_size else 0 return response.json(0, {'icloudList': icloud_list, 'notShow': not_show}) except Exception as e: LOGGER.info('获取云盘播放列表异常:{}'.format(repr(e))) return response.json(500) @staticmethod def icloud_play(request_dict, response): """ 云盘播放 @param request_dict: 请求参数 @request_dict time_stamp: 时间戳 @request_dict uid: 设备uid @request_dict channel: 设备通道 @param response: 响应对象 @return: response """ LOGGER.info('播放云盘记录') time_stamp = request_dict.get('time_stamp', None) uid_token = request_dict.get('uid', None) channel = request_dict.get('channel', None) file_type = request_dict.get('file_type', None) if not all([time_stamp, uid_token, channel, file_type]): return response.json(444, {'error param': 'time_stamp, uid_token, channel, file_type'}) if file_type != '1': return response.json(444, {'error param': 'file_type'}) token = jwt.decode(uid_token, OAUTH_ACCESS_TOKEN_SECRET, algorithms='HS256') uid = token.get('uid', '') user_id = token.get('user_id', '') icloud_record_qs = IcloudStorageRecord.objects.filter(user_id=user_id, uid=uid, channel=channel, file_type=file_type, time_stamp=time_stamp).values('bucket_id', 'fg') if not icloud_record_qs.exists(): return response.json(173) bucket_id = icloud_record_qs[0]['bucket_id'] fg = int(icloud_record_qs[0]['fg']) bucket_qs = VodBucketModel.objects.filter(id=bucket_id).values('bucket', 'region', 'mold') if not bucket_qs.exists(): return response.json(173) bucket_name = bucket_qs[0]['bucket'] bucket_region = bucket_qs[0]['region'] mold = bucket_qs[0]['mold'] try: s3_obj = AmazonS3Util( AWS_ACCESS_KEY_ID[mold], AWS_SECRET_ACCESS_KEY[mold], bucket_region ) playlist_entries = [] for i in range(15): shift = (i + 1) * 4 duration = (fg >> shift) & 0xf if duration > 0: ts_file = '{user_id}/ts_file/{uid}/channel{channel}/{time_stamp}/ts{i}.ts'.format( user_id=user_id, uid=uid, channel=channel, time_stamp=time_stamp, i=i) response_url = s3_obj.generate_file_obj_url(bucket_name, ts_file) playlist_entries.append({ 'name': response_url, 'duration': duration, }) playlist = PlaylistGenerator(playlist_entries).generate() response = HttpResponse(playlist) response['Content-Type'] = 'application/octet-stream' response['Content-Disposition'] = 'attachment;filename="play.m3u8"' return response except Exception as e: LOGGER.info('云盘播放异常:{}'.format(repr(e))) return response.json(500) @staticmethod def do_change_icloud_status(request_dict, response, user_id): """ 修改云盘开关 @param request_dict: 请求参数 @param user_id: 用户id @request_dict status: 开关状态, 0: 关闭, 1: 开启 @param response: 响应对象 @return: response """ status = request_dict.get('status', None) if not status: return response.json(444, 'status') try: device_user_qs = Device_User.objects.filter(userID=user_id) if not device_user_qs.exists(): return response.json(173) icloud_use_details_qs = IcloudUseDetails.objects.filter(user_id=user_id) icloud_use_details_qs.update(detect_status=status) return response.json(0) except Exception as e: print(e) return response.json(500)