# -*- encoding: utf-8 -*- """ @File : AgentDeviceController.py @Time : 2024/3/8 13:55 @Author : stephen @Email : zhangdongming@asj6.wecom.work @Software: PyCharm """ import os import csv import time from datetime import datetime import calendar from dateutil.relativedelta import relativedelta from collections import defaultdict from decimal import Decimal import traceback import threading from django.db import transaction from django.db.models import Q from django.http import QueryDict from django.views import View from django.core.paginator import Paginator from AgentModel.models import AgentCustomerInfo, AgentDeviceOrder, AgentDevice, AgentCloudServicePackage, CustomUIDPool, \ DeviceCustomUID from Model.models import DeviceTypeModel from Object.ResponseObject import ResponseObject from Ansjer.config import LOGGER from Object.TokenObject import TokenObject class AgentDeviceView(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 delete(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') delete = QueryDict(request.body) if not delete: delete = request.GET return self.validation(delete, request, operation) def put(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') put = QueryDict(request.body) return self.validation(put, request, operation) def validation(self, request_dict, request, operation): language = request_dict.get('language', 'en') response = ResponseObject(language, 'pc') # 订单结算界面 if operation == 'XXXXX': pass else: tko = TokenObject( request.META.get('HTTP_AUTHORIZATION'), returntpye='pc') if tko.code != 0: return response.json(tko.code) response.lang = tko.lang userID = tko.userID if operation == 'getAgentDevice': return self.get_agent_device(userID, request_dict, response) elif operation == 'getAgentDeviceOrder': return self.get_agent_device_order(userID, request_dict, response) elif operation == 'batchBandDevice': return self.batch_band_device(userID, request, request_dict, response) elif operation == 'customUidBind': return self.custom_uid_bindings(request_dict, response) else: return response.json(444, 'operation') def get_agent_device(self, userID, request_dict, response): """ 查询设备明细 @param userID: userID @param request_dict: 请求参数 @param request_dict ac_id: 代理商id @param request_dict device_name: 设备名字 @param request_dict status: 设备类型 @param request_dict serial_number: 设备9位序列号 @param response: 响应对象 @return: """ device_name = request_dict.get('device_name', None) status = request_dict.get('status', None) serial_number = request_dict.get('serial_number', None) page = int(request_dict.get('page', 1)) # 默认为第一页 page_size = int(request_dict.get('page_size', 10)) # 默认每页10条记录 try: agent_customer_info = AgentCustomerInfo.objects.filter(user_id=userID).first() if agent_customer_info is None: agent_device_qs = AgentDevice.objects.order_by('ac_id', '-created_time') else: ac_id = agent_customer_info.id agent_device_qs = AgentDevice.objects.filter(ac_id=ac_id).order_by('ac_id', '-created_time') if device_name: # 根据device_name查询对应的type值 device_types = list(DeviceTypeModel.objects.filter(name=device_name).values_list('type', flat=True)) agent_device_qs = agent_device_qs.filter(type__in=device_types) if status: agent_device_qs = agent_device_qs.filter(status=status) if serial_number: agent_device_qs = agent_device_qs.filter(serial_number=serial_number) # 应用分页 paginator = Paginator(agent_device_qs, page_size) current_page = paginator.get_page(page) # 构造返回列表 device_list = [] for device in current_page: device_type = DeviceTypeModel.objects.filter(type=device.type).first() device_name = device_type.name if device_type else device.type agent_customer_info = AgentCustomerInfo.objects.filter(id=device.ac_id).first() company_name = agent_customer_info.company_name if agent_customer_info else device.ac_id device_list.append({ 'id': device.id, 'ac_id': device.ac_id, 'company_name': company_name, 'status': device.status, 'serial_number': device.serial_number, 'device_name': device_name, 'at_time': device.at_time, }) # 包含分页信息的响应 response_data = { 'list': device_list, 'total': paginator.count, 'page': current_page.number, 'page_size': page_size, 'num_pages': paginator.num_pages, } return response.json(0, response_data) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) def calculate_profit_or_revenue(self, agent_device_orders, package_details, time_unit, metric_type, start_time, end_time): """ 计算利润或者营业额 @param agent_device_orders: 代理设备订单 @param package_details: 代理套餐详情 @param time_unit: 时间单位 @param metric_type: 利润或者营业额 @param start_time: 开始时间 @param end_time: 结束时间 @return: """ summary = defaultdict(lambda: {"云存": Decimal('0.00'), "4G": Decimal('0.00'), "all": Decimal('0.00')}) time_format = { "month": "%Y-%m", "year": "%Y", "quarter": lambda x: f"{x.year}年{(x.month - 1) // 3 + 1}季度" } for order in agent_device_orders: package = package_details.get(order.csp_id) if not package: continue # 根据利润类型计算利润或者直接使用营业额 if metric_type == 1: # 利润 profit = order.profit else: # 营业额 profit = order.profit_amount # 区分云服务 + 4G套餐并加入 summary service_type = "云存" if package.type == 1 else "4G" time_key = datetime.fromtimestamp(order.created_time).strftime( time_format[time_unit]) if time_unit != "quarter" else time_format[time_unit]( datetime.fromtimestamp(order.created_time)) summary[time_key][service_type] += profit summary[time_key]["all"] += profit # 补全时间段内所有可能的时间单位 current_time = start_time while current_time < end_time: time_key = current_time.strftime(time_format[time_unit]) if time_unit != "quarter" else time_format[ time_unit](current_time) if time_key not in summary: summary[time_key] = {"云存": Decimal('0.00'), "4G": Decimal('0.00'), "all": Decimal('0.00')} current_time += relativedelta(months=1) if time_unit == "month" else relativedelta( years=1) if time_unit == "year" else relativedelta(months=3) return [{"time": time, **data} for time, data in sorted(summary.items())] def get_agent_device_order(self, userID, request_dict, response): """ 查询设备订单明细 @param userID: userID @param request_dict: 请求参数 @param request_dict startTime: 开始时间 @param request_dict endTime: 结束时间 @param request_dict timeUnit: 时间单位 @param request_dict metric_type: 利润或者营业额 @param response: 响应对象 @return: """ try: startTime = int(request_dict.get('startTime', 1704038400)) endTime = int(request_dict.get('endTime', 1732982400)) timeUnit = request_dict.get('timeUnit', 'month') metric_type = int(request_dict.get('metric_type', 0)) # endTime变成每个月最后一天 end_datetime = datetime.fromtimestamp(endTime) month_str = end_datetime.strftime('%Y-%m') year, month = int(month_str.split('-')[0]), int(month_str.split('-')[1]) end = calendar.monthrange(year, month)[1] end_timestamp = datetime(year, month, end, 23, 59, 59).timestamp() endTime = int(end_timestamp) agent_customer_info = AgentCustomerInfo.objects.filter(user_id=userID).first() if not agent_customer_info: return response.json(104, 'Agent customer not found') agent_device_orders = AgentDeviceOrder.objects.filter( ac_id=agent_customer_info.id, created_time__gte=startTime, created_time__lte=endTime, status__in=[1, 2] ) # 获取代理套餐包id package_ids = agent_device_orders.values_list('csp_id', flat=True).distinct() package_details = {pkg.id: pkg for pkg in AgentCloudServicePackage.objects.filter(id__in=package_ids)} start_time = datetime.fromtimestamp(startTime) end_time = datetime.fromtimestamp(endTime) result = self.calculate_profit_or_revenue(agent_device_orders, package_details, timeUnit, metric_type, start_time, end_time) total_4G = Decimal('0.00') total_cloud = Decimal('0.00') # 遍历result列表来累加4G和云存的值 for item in result: total_4G = item['4G'] + total_4G total_cloud = item['云存'] + total_cloud response_data = { "list": result, "total_4G": total_4G, # 4G的总和 "total_云存": total_cloud, # 云存的总和 } return response.json(0, response_data) except Exception as e: error_msg = f"error_line:{traceback.format_exc()}, error_msg:{str(e)}" return response.json(500, error_msg) def agent_devices_from_csv(self, ac_id, device_type, userID, file_path): """ 异步批量绑定设备 """ try: with open(file_path, 'r') as file: reader = csv.DictReader(file) devices_to_create = [] # 先收集所有CSV中的序列号 csv_serial_numbers = [row.get('serial_number') for row in reader] # 去重 unique_serial_numbers = set(csv_serial_numbers) existing = set(AgentDevice.objects.filter( serial_number__in=unique_serial_numbers ).values_list('serial_number', flat=True)) for row in unique_serial_numbers: serial_number = row.get('serial_number') if serial_number not in existing: device = AgentDevice( ac_id=ac_id, serial_number=serial_number, type=device_type, status=0, created_time=int(time.time()), created_by=userID, updated_time=int(time.time()), updated_by=userID ) devices_to_create.append(device) # 使用Django的bulk_create来批量创建对象 if devices_to_create: with transaction.atomic(): AgentDevice.objects.bulk_create(devices_to_create, batch_size=200) # 删除文件 os.remove(file_path) except Exception as e: LOGGER.info('errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) def batch_band_device(self, userID, request, request_dict, response): """ 批量绑定设备 @param ac_id: ac_id 代理商id @param userID: userID @param csv_file_path: 文件路径 @param response: 响应对象 @return: """ ac_id = request_dict.get('ac_id', None) device_name = request_dict.get('device_name', None) csv_file = request.FILES['file'] upload_dir = os.path.join('static', 'uploaded_files') if not all([ac_id, device_name]): return response.json(444) try: device_type_dict = DeviceTypeModel.objects.filter(name=device_name).values('type').first() device_type = device_type_dict['type'] if not os.path.exists(upload_dir): os.makedirs(upload_dir) file_path = os.path.join(upload_dir, csv_file.name) with open(file_path, 'wb+') as destination: for chunk in csv_file.chunks(): destination.write(chunk) # 创建并启动线程来异步执行任务 thread = threading.Thread(target=self.agent_devices_from_csv, args=(ac_id, device_type, userID, file_path)) thread.start() return response.json(0) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @classmethod def device_binding_or_unbinding(cls, serial_number, bind_type): """ 设备绑定或解绑 @param serial_number: 设备序列号 @param bind_type: 绑定类型 1:绑定 2:解绑 @return: 无返回值 """ try: # 获取设备信息 device_info = AgentDevice.objects.filter(serial_number=serial_number) if not device_info.exists(): return LOGGER.info('同步更新设备激活状态serial:{},type:{}'.format(serial_number, bind_type)) n_time = int(time.time()) if bind_type == 1: # 绑定设备 device_info.update(status=1, updated_time=n_time, at_time=n_time) elif bind_type == 2: # 解绑设备 device_info.update(status=0, updated_time=n_time) except Exception as e: LOGGER.error('*****AgentDeviceView.device_binding_or_unbinding:errLine:{}, errMsg:{}' .format(e.__traceback__.tb_lineno, repr(e))) @classmethod def custom_uid_bindings(cls, request_dict, response): try: uid = request_dict.get('uid', None) customer_name = request_dict.get('customer_name', None) status = request_dict.get('status', None) device_mac = request_dict.get('device_mac', None) page = int(request_dict.get('page', 1)) page_size = int(request_dict.get('pageSize', 20)) filters = Q() if uid: filters &= Q(uid__icontains=uid) if customer_name: filters &= Q(customer_name__icontains=customer_name) if status is not None: filters &= Q(status=status) if device_mac: bound_uids = DeviceCustomUID.objects.filter( device_mac__icontains=device_mac ).values_list('uid', flat=True) filters &= Q(uid__in=list(bound_uids)) custom_uid_pool_qs = CustomUIDPool.objects.filter(filters).order_by('-updated_time') paginator = Paginator(custom_uid_pool_qs, page_size) page_obj = paginator.page(page) uid_list = [obj.uid for obj in page_obj] bindings = DeviceCustomUID.objects.filter(uid__in=uid_list) # 构建 uid -> 多条绑定记录 map binding_map = {} for b in bindings: binding_map.setdefault(b.uid, []).append(b) # 构建结果列表 result_list = [] for obj in page_obj: uid_bindings = binding_map.get(obj.uid, []) if uid_bindings: for bind in uid_bindings: result_list.append({ 'id': obj.id, 'uid': obj.uid, 'type': obj.type, 'customer_name': obj.customer_name, 'uid_status': obj.status, 'created_time': obj.created_time, 'updated_time': obj.updated_time, 'device_mac': bind.device_mac, 'device_status': bind.status, 'bind_time': bind.created_time, }) else: result_list.append({ 'id': obj.id, 'uid': obj.uid, 'type': obj.type, 'customer_name': obj.customer_name, 'uid_status': obj.status, 'created_time': obj.created_time, 'updated_time': obj.updated_time, 'device_mac': '', 'device_status': None, 'bind_time': None, }) return response.json(0, { 'list': result_list, 'total': paginator.count }) except Exception as e: print(e) return response.json(500, f'error_line:{e.__traceback__.tb_lineno}, error_msg:{repr(e)}')