import copy import logging import time import traceback import uuid from itertools import islice import openpyxl from django.core.exceptions import ObjectDoesNotExist from django.db import transaction from django.db.models import Q from django.http import FileResponse from django_filters.rest_framework import DjangoFilterBackend from openpyxl.comments import Comment from openpyxl.worksheet.cell_range import CellRange from openpyxl.worksheet.datavalidation import DataValidation from rest_framework.decorators import action from rest_framework.filters import SearchFilter, OrderingFilter from rest_framework.permissions import IsAuthenticated from rest_framework.views import APIView from rest_framework_jwt.authentication import JSONWebTokenAuthentication from ChaCeRndTrans import settings from ChaCeRndTrans.basic import CCAIResponse from ChaCeRndTrans.code import * from common.models import DataDictionaryDetail, DataDictionary from common.serializers.dict_serializer import DictDetailSerializer from rbac.models import Company, UserProfile from rbac.serializers.user_serializer import CompanySerializer from utils.custom import CustomViewBase, CommonPagination, RbacPermission, req_operate_by_user, asyncDeleteFile, \ ErrorTable, MyCustomError, time_int_to_str, time_str_to_int err_logger = logging.getLogger('error') class CompanyCustomViewSet(CustomViewBase): ''' 公司管理 ''' perms_map = ( {"*": "admin"}, {"*": "company_all"}, {"get": "company_list"}, {"post": "company_create"}, {"put": "company_edit"}, {"delete": "company_delete"} ) queryset = Company.objects.all() serializer_class = CompanySerializer pagination_class = CommonPagination filter_backends = (DjangoFilterBackend, SearchFilter, OrderingFilter) # filter_fields = ("",) search_fields = ("name", "EUCC") ordering_fields = ("id",) authentication_classes = (JSONWebTokenAuthentication,) permission_classes = (RbacPermission,) def list(self, request, format=None): pagination = {} try: params = request.GET keyword = params.get('keyword') page_size = params.get('size') page = params.get('page') sort = params.get('sort') order_by = 'N.id desc' if sort: order_by = sort if page is None: page = 1 if page_size is None: page_size = 10 start_index = (int(page) - 1) * int(page_size) where = 'WHERE 1=1 ' param = [] if keyword: where += 'AND (N.name LIKE %s OR N.EUCC LIKE %s ) ' param.append('%' + keyword + '%') param.append('%' + keyword + '%') sql = 'SELECT N.id, N.name, N.MainId, userMid, U.name as user, U.username,' \ 'EUCC, employeesCount, hightechCode, hightechDate, AmStart, AmEnd, PmStart, PmEnd, tech, tolerance, checkIn ' \ 'FROM chace_rnd.rbac_company AS N ' \ 'LEFT JOIN chace_rnd.rbac_userprofile as U ON U.MainId = N.userMid ' \ '%s ORDER BY %s LIMIT %s,%s ' % (where, order_by, start_index, page_size) query_rows = Company.objects.raw(sql, param) count_sql = """ select N.id, count(N.id) as count from chace_rnd.rbac_company AS N %s """ % where count_result = Company.objects.raw(count_sql, param) count = 0 if len(count_result) > 0: count = count_result[0].count rows = [] for item in query_rows: item.__dict__.pop('_state') if item.__dict__['tech'] and item.__dict__['tech'] != '': item.__dict__['tech'] = item.__dict__['tech'].split(',') if not item.__dict__['user'] or item.__dict__['user'] == '': item.__dict__['user'] = item.__dict__['username'] item.__dict__['tolerance'] = time_int_to_str(item.__dict__['tolerance']) rows.append(item.__dict__) pagination = { "page": page, "page_size": page_size } return CCAIResponse(rows, count=count, pagination=pagination) except Exception as e: err_logger.error("user: %s, get company list failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取公司列表失败", SERVER_ERROR) @action(methods=["get"], detail=True, permission_classes=[IsAuthenticated], url_path="getAssociationUser", url_name="getAssociationUser") def getAssociationUser(self, request, pk=None): """ 获取该公司关联的用户列表 """ try: if not pk: return CCAIResponse("请选择对应公司", BAD) instance = Company.objects.get(id=int(pk)) user_list = UserProfile.objects.filter(company=instance).values() return CCAIResponse(user_list) except Company.DoesNotExist: return CCAIResponse("该公司已被删除！", BAD) except Exception as e: err_logger.error("user: %s, get getAssociationUser list failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取该公司关联的用户列表失败", SERVER_ERROR) def create(self, request, *args, **kwargs): """ 直接新增公司 """ try: data = req_operate_by_user(request) # 用户选择为其添加公司的用户 userMid = data.get('userMid') if not userMid: return CCAIResponse("Missing User Information", BAD) user = UserProfile.objects.filter(MainId=userMid).first() if not user: return CCAIResponse("User Information Error", BAD) companyMid = uuid.uuid4().__str__() data['MainId'] = companyMid tech = data.get('tech') # 技术领域 if tech: data['tech'] = ','.join(tech) tolerance = data.get('tolerance') # 容错时间 if tolerance: data['tolerance'] = time_str_to_int(tolerance) serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) with transaction.atomic(): company = self.perform_create(serializer) user.company.add(company) # headers = self.get_success_headers(serializer.data) return CCAIResponse(data="success") except Exception as e: err_logger.error("user: %s, create company failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("直接新增公司失败", SERVER_ERROR) def perform_create(self, serializer): # 重写序列化创建,返回创建的公司对象 return serializer.save() def update(self, request, *args, **kwargs): """ 如果用户修改拥有者 """ try: data = req_operate_by_user(request) tech = data.get('tech') tolerance = data.get('tolerance') # 容错时间 userMid = data.get('userMid') if tech: data['tech'] = ','.join(str(x) for x in tech) if tolerance: data['tolerance'] = time_str_to_int(tolerance) partial = kwargs.pop('partial', False) # True:所有字段全部更新， False:仅更新提供的字段 instance = self.get_object() with transaction.atomic(): # 检查是否更改拥有者 flag = False if userMid and instance.userMid != userMid: flag = True new_user = UserProfile.objects.filter(MainId=userMid).select_for_update().first() # 新的拥有者 old_user = UserProfile.objects.filter(MainId=instance.userMid).select_for_update().first() # 旧的拥有者 serializer = self.get_serializer(instance, data=data, partial=partial) serializer.is_valid(raise_exception=True) self.perform_update(serializer) if flag: if not new_user: transaction.set_rollback(True) return CCAIResponse("用户已被删除!", BAD) else: new_user.company.add(instance) if old_user: old_user.company.remove(instance) if getattr(instance, '_prefetched_objects_cache', None): # If 'prefetch_related' has been applied to a queryset, we need to # forcibly invalidate the prefetch cache on the instance. instance._prefetched_objects_cache = {} return CCAIResponse(data="success") except Exception as e: err_logger.error("user: %s, update company Info failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("修改公司信息失败列表失败", SERVER_ERROR) @action(methods=["get"], detail=False, permission_classes=[IsAuthenticated], url_path="getActiveUser", url_name="getActiveUser") def getActiveUser(self, request): """ 获取主账号可用用户列表,供公司拥有者选择 """ try: users = UserProfile.objects.filter(is_sub=2, is_active=1, MainId__isnull=False).values("id", "name", "username", "MainId") return CCAIResponse(users) except Exception as e: err_logger.error("user: %s, getActiveUser failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取可选择的公司拥有者列表失败", SERVER_ERROR) @action(methods=["get"], detail=False, permission_classes=[IsAuthenticated], url_path="getCompanyList", url_name="getCompanyList") def getCompanyList(self, request): """ 后台用户获得公司列表 """ try: if 2 == request.user.label: companies = Company.objects.all().values('id', 'name') result = [item for item in companies] else: return CCAIResponse() return CCAIResponse(result) except Exception as e: err_logger.error("user: %s, getUserCompanyInfo failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取用户公司信息失败", SERVER_ERROR) @action(methods=["get"], detail=False, permission_classes=[IsAuthenticated], url_path="getMainUserList", url_name="getMainUserList") def getMainUserList(self, request): """ 获取主账户id,username,name,MainId """ try: keyword = request.GET.get('keyword') query = Q(is_sub=2) if keyword: query &= (Q(username__icontains=keyword) | Q(name__icontains=keyword)) mainUser = UserProfile.objects.filter(query).values('id', 'username', 'name', 'MainId') return CCAIResponse(mainUser) except Exception as e: err_logger.error("user: %s, getMainUser list failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取主用户列表失败", SERVER_ERROR) @action(methods=["get"], detail=False, permission_classes=[IsAuthenticated], url_path="getUserCompanyInfo", url_name="getUserCompanyInfo") def getUserCompanyInfo(self, request): try: params = request.GET companyMid = params.get('companyMid') query = Q(MainId=companyMid) try: company = Company.objects.get(query) except ObjectDoesNotExist: return CCAIResponse('缺少公司参数', BAD) data = CompanySerializer(company).data if data.get('tech'): data['tech'] = [int(x) for x in data['tech'].split(',')] if 'tolerance' in data: data['tolerance'] = time_int_to_str(data['tolerance']) return CCAIResponse(data) except Exception as e: err_logger.error("user: %s, getUserCompanyInfo failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取用户公司信息失败", SERVER_ERROR) @action(methods=["get"], detail=False, permission_classes=[IsAuthenticated], url_path="getTechnologyTree", url_name="getTechnologyTree") def getTechnologyTree(self, request): """ 技术领域树 """ try: dataDictionary = DataDictionary.objects.get(DictionaryCode='TechArea') dataDictionaryDetail = DataDictionaryDetail.objects.filter( DataDictionaryId=dataDictionary.DataDictionaryId).values('id', 'DataDictionaryDetailId', 'ParentId', 'DictionaryCode', 'DictionaryValue', 'ParentCode', 'FullName') data = DictDetailSerializer(dataDictionaryDetail, many=True) tree_dict = {} for item in data.data: if item["parent_id"] is None: item["children"] = [] top_permission = copy.deepcopy(item) tree_dict[item["dict_detail_id"]] = top_permission else: children_permission = copy.deepcopy(item) tree_dict[item["dict_detail_id"]] = children_permission tree_data = [] for i in tree_dict: if tree_dict[i]["parent_id"]: pid = tree_dict[i]["parent_id"] parent = tree_dict[pid] parent.setdefault("children", []).append(tree_dict[i]) # from operator import itemgetter # parent["children"] = sorted(parent["children"], key=itemgetter("id")) else: tree_data.append(tree_dict[i]) return CCAIResponse(tree_data, status=OK) except Exception as e: err_logger.error("user: %s, getUserCompanyInfo failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("获取用户公司信息失败", SERVER_ERROR) @action(methods=["get"], detail=False, permission_classes=[IsAuthenticated], url_path="download", url_name="download") def download(self, request): """ 导入模板下载 """ try: # 查询公司的项目列表 companymid = request.GET.get('companyMid') if not companymid: return CCAIResponse('Missing or invalid company information', BAD) company = Company.objects.filter(MainId=companymid).first() if not company: return CCAIResponse('Company param error', BAD) users = UserProfile.objects.filter(is_sub=2, is_active=1, MainId__isnull=False).values("id", "name", "username", "MainId") username_user = [str((item['name'] if item['name'] else '') + '（' + item['username'] + '）') for item in users] # 使用用户姓名(账号) 做唯一 title_data = ["*公司名称", "*拥有者", "企业统一信用代码", "高企编码", "企业员工总数"] name = str(int(time.time() * 10000)) # 时间戳命名 workbook = openpyxl.Workbook() worksheet = workbook.active worksheet.title = "公司导入模板" # 添加隐藏的选项参考表 worksheet2 = workbook.create_sheet("选项参考表") worksheet2.sheet_state = "hidden" # 将工作表隐藏起来 worksheet2.cell(1, 1, '拥有者参考') # 写入拥有者参考数据 for idx, project in enumerate(username_user, start=2): # 从第二行开始写入数据 worksheet2.cell(idx, 1, str(project)) for index, item in enumerate(title_data, start=1): worksheet.cell(1, index, item) worksheet['A1'].comment = Comment("必填项，请输入正确的公司名称！", 'Author') worksheet['B1'].comment = Comment("必填项，请从提供的选项中选择！", 'Author') # 创建一个数据有效性验证对象 usersDv = DataValidation(type="list", formula1=f'=选项参考表!$A$2:$A${len(username_user) + 1}', showErrorMessage=True, errorTitle="输入错误", error="请从给定的选项中选择") # 数据有效性验证应用 worksheet.add_data_validation(usersDv) usersDv.add(CellRange("B2:B10000")) # 设置第一列的数据格式为文本类型 worksheet.column_dimensions['A'].number_format = '@' worksheet.column_dimensions['C'].number_format = '@' worksheet.column_dimensions['D'].number_format = '@' # worksheet.cell(2, 10).value = '=H2*I2' # for row in range(2, 10000): # worksheet.cell(row, 10).value = '=IF(ISBLANK(H{}) OR ISBLANK(I{}), "", H{}*I{})'.format(row, row, row, row) workbook.save(filename=name + ".xlsx") fileName = name + ".xlsx" file = open(fileName, 'rb') s_name = "公司导入模板.xlsx" response = FileResponse(file) response['Content-Type'] = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' # 规定返回xlsx response['Content-Disposition'] = f'attachment;filename="{s_name}"' response['file_name'] = fileName # 异步删除文件 threadPool.submit(asyncDeleteFile, request, fileName) return response except Exception as e: err_logger.error("user: %s, get company template file failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("下载公司导入模板失败", SERVER_ERROR) @action(methods=["post"], detail=False, permission_classes=[IsAuthenticated], url_path="import_excel", url_name="import_excel") def import_excel(self, request): """导入excel表数据""" try: data = req_operate_by_user(request) file_name = data.get('file_name') if not file_name: return CCAIResponse('Missing params', BAD) users = UserProfile.objects.filter(is_sub=2, is_active=1, MainId__isnull=False).values("id", "name", "username", "MainId") username_user = {str((item['name'] if item['name'] else '') + '（' + item['username'] + '）'): item['MainId'] for item in users} # 使用用户姓名(账号) 做唯一 company_list = [] excel_file_list = [] file_paths = file_name.split(",") for key in range(len(file_paths)): if file_paths[key] == '' or file_paths[key] is None: continue excel_file = file_paths[key].replace(settings.SHOW_UPLOAD_PATH, settings.FILE_PATH) excel_file_list.append(excel_file) # 打开工作文件 workbook = openpyxl.load_workbook(excel_file, data_only=True) table = workbook['公司导入模板'] # table = workbook.active rows = table.max_row # 总行数 check_value = [table.cell(row=1, column=i).value for i in range(1, 4)] # 获取第一行的值 if check_value[0] != '*公司名称' and check_value[1] != '*拥有者' and check_value[2] != '*企业统一信用代码': return CCAIResponse("文件不符合模板标准", SERVER_ERROR) user_data = [] # 项目编号列 for row in islice(table.iter_rows(values_only=True), 1, None): if row[1]: user_data.append(row[0]) # 项目编号列 try: columns = ["companyName", "userName", "EUCC", "hightechCode", "employeesCount"] columns_name = ["企业名称", "拥有者", "企业同一信用代码", "高企编码", "员工人数"] errorTable = ErrorTable(columns) # 循环外层控制事务,因为物料记录的创建依赖与项目分摊的创建 with transaction.atomic(): for rindex, row in enumerate(islice(table.iter_rows(values_only=True), 1, None)): # 跳过空白隐藏行 if all(cell is None for cell in row): continue row_value = [r for r in row] check_index = [0, 1,] # 校验必填项 skip = False for index in check_index: # 前2项为必填项 if row[index] is None or row[index] == '': skip = True errorTable.add_one_row(row_value, f'{columns_name[index]}为必填项') break if skip: skip = False continue if str(row[1]) not in username_user: errorTable.add_one_row(row_value, f'用户错误') continue # 获取对应的用户 userMid = username_user[str(row[1])] try: user = UserProfile.objects.get(MainId=userMid) except UserProfile.DoesNotExist: errorTable.add_one_row(row_value, f'用户错误') continue # 新增公司 companyMid = uuid.uuid4().__str__() # 公司mid company = Company() company.name = str(row_value[0]) company.MainId = companyMid company.userMid = userMid if row_value[2]: company.EUCC = str(row_value[2]) if row_value[3]: company.hightechCode = str(row_value[3]) company.employeesCount = int(row_value[4]) if row_value[4] else 0 company.save() user.company.add(company) if errorTable.has_data(): # 存在错误信息,回滚并返回所有错误行的提示信息 transaction.set_rollback(True) return CCAIResponse(errorTable.get_table(), 200) except MyCustomError as e: return CCAIResponse(f'文件读取第{rindex}行数据多于标题列', BAD) except Exception as e: err_logger.error("user: %s, 文件路径:%s 记录公司导入错误日志: \n%s" % (request.user.id, excel_file, traceback.format_exc())) return CCAIResponse('数据有误,请检查后重新导入', BAD) try: for file_path in excel_file_list: # 异步删除文件 threadPool.submit(asyncDeleteFile, request, file_path) # if os.path.exists(file_path): # 如果文件存在 # # 删除文件，可使用以下两种方法。 # os.remove(file_path) # # 删除空目录，不是空目录时候rmdir不会删除 # # try: # # os.rmdir(file_path) # # except Exception as e: # # logger.error("user: %s, rmdir highTech failed: \n%s" % (request.user.id, traceback.format_exc())) except Exception as e: err_logger.error("user: %s, import_excel view create failed: \n%s" % ( request.user.id, traceback.format_exc())) return CCAIResponse("导入存入失败", SERVER_ERROR) return CCAIResponse(data="success") except Exception as e: err_logger.error("user: %s, import aigc file failed: \n%s" % (request.user.id, traceback.format_exc())) return CCAIResponse("导入公司名单失败", SERVER_ERROR) class CompanyNameListAPIView(APIView): authentication_classes = (JSONWebTokenAuthentication,) permission_classes = (IsAuthenticated,) ''' 获取公司名称列表 ''' # throttle_classes = (AnonRateThrottle, UserRateThrottle) def get(self, request, format=None): blurry_company_info_req = '' try: params = request.GET company_name = params.get('company_name') rows = [] qixinbao_failed = False if company_name: blurry_company_info_req = getBlurryCompanyInfo(request, company_name, 0) if blurry_company_info_req: base_info = blurry_company_info_req.json() if base_info['status'] == '200': companys = base_info['data']['items'] for company in companys: dict = {'value': company['name']} rows.append(dict) # rows = [ # {'value':"深圳市骏鼎达新材料股份有限公司"}, # {'value':"东莞市骏鼎达新材料科技有限公司"}, # {'value':"昆山骏鼎达电子科技有限公司"}, # {'value':"苏州骏鼎达新材料科技有限公司"}, # {'value':"江门骏鼎达新材料科技有限公司"}, # {'value':"深圳市骏鼎达新材料股份有限公司重庆分公司"}, # {'value':"深圳市骏鼎达新材料股份有限公司武汉分公司"}, # {'value':"龙川县骏鼎达新材料有限公司"}, # {'value':"佛山市骏鼎达五金有限公司"}, # {'value':"駿鼎達國際有限公司"} # ] return CCAIResponse(rows) except Exception as e: logger.error("user: %s, get blurry company failed: %s, detail: %s" % (request.user.id, traceback.format_exc(), blurry_company_info_req)) return CCAIResponse("获取地区失败", SERVER_ERROR)