from django.shortcuts import render from rest_framework.mixins import ListModelMixin from rest_framework.permissions import IsAuthenticated from rest_framework.viewsets import GenericViewSet, ModelViewSet from .models import * from .serializers import * from rest_framework.decorators import action from django.conf import settings from rest_framework import status from rest_framework.response import Response import zipfile import rarfile from apps.system.models import Organization from openpyxl import Workbook, load_workbook from django.db.models import Count from utils.pagination import PageOrNot from rest_framework.views import APIView from apps.supervision.models import Content, Record from apps.supervision.serializers import ContentSerializer, RecordCreateSerializer, RecordSerializer from apps.system.mixins import CreateUpdateCustomMixin from utils.queryset import get_child_queryset2 from apps.system.permission_data import RbacFilterSet from django.utils import timezone from apps.supervision.permission import RecordPermission from django.utils.decorators import method_decorator from django.views.decorators.cache import cache_page # Create your views here. import json class RecordMixin: def perform_authentication(self, request): """ Perform authentication on the incoming request. Note that if you override this and simply 'pass', then authentication will instead be performed lazily, the first time either `request.user` or `request.auth` is accessed. """ if request.user and request.method == "GET" and request.user.username != "admin": QueryRecord.objects.create( user=request.user, path=request.path, ip=request.META.get("HTTP_X_FORWARDED_FOR"), method=request.method, search=request.query_params.get("search", None), query=request.query_params ) class AbilityContentViewSet(CreateUpdateCustomMixin, ModelViewSet): """ 资质能力报送清单：增删改查 """ perms_map = {"get": "*", "post": "content", "put": "content", "delete": "content"} queryset = Content.objects.filter(cate=2) serializer_class = ContentSerializer pagination_class = None search_fields = ["name", "desc"] filterset_fields = ["type", "cate"] ordering = ["sortnum", "create_time"] def perform_create(self, serializer): if hasattr(self.queryset.model, "belong_dept"): serializer.save(create_by=self.request.user, belong_dept=self.request.user.dept, cate=2) else: serializer.save(create_by=self.request.user, cate=2) def perform_update(self, serializer): serializer.save(update_by=self.request.user) class AbilityRecordViewSet(RbacFilterSet, PageOrNot, CreateUpdateCustomMixin, ModelViewSet): perms_map = {"get": "*", "post": "*", "put": "*", "delete": "*"} queryset = Record.objects.filter(content__cate=2) serializer_class = RecordSerializer search_fields = ["content__name"] ordering_fields = ["content__sortnum", "belong_dept__sort"] ordering = ["-task", "content__sortnum", "-create_time"] filterset_fields = ["content", "content__cate", "belong_dept", "state"] def filter_queryset(self, queryset): if not self.request.query_params.get("pageoff", None): queryset = queryset.exclude(state="待发布") return super().filter_queryset(queryset) def get_serializer_class(self): if self.action == "create": return RecordCreateSerializer return super().get_serializer_class() @action(methods=["put"], detail=True, perms_map={"put": "record_up"}) def up(self, request, *args, **kwargs): """ 报送 """ obj = self.get_object() if obj.state in ["待整改", "待报送", "已报送"]: obj.files.clear() obj.files.add(*request.data["files"]) obj.noteb = request.data["noteb"] obj.state = "已报送" obj.up_user = request.user obj.up_date = timezone.now() obj.save() return Response(status=status.HTTP_200_OK) return Response("记录状态错误", status=status.HTTP_400_BAD_REQUEST) def create(self, request, *args, **kwargs): """ 主动报送 """ data = request.data if data.get("files", None): serializer = RecordCreateSerializer(data=data) serializer.is_valid(raise_exception=True) content = Content.objects.get(pk=data["content"]) if content.cate == 2: sdata = { "belong_dept": request.user.dept, "is_self": True, "content_name": content.name, "content_desc": content.desc, "state": "已报送", "up_user": request.user, "up_date": timezone.now(), } serializer.save(**sdata) return Response(status=status.HTTP_201_CREATED) return Response("该材料不是资质能力!", status=status.HTTP_400_BAD_REQUEST) return Response("未上传文件", status=status.HTTP_400_BAD_REQUEST) @action(methods=["put"], detail=True, perms_map={"put": "record_reject"}, permission_classes=[RecordPermission]) def reject(self, request, *args, **kwargs): """ 驳回 """ obj = self.get_object() if obj.state == "已报送": if request.data.get("opinion", None): obj.opinion = request.data["opinion"] obj.state = "待整改" obj.save() return Response(status=status.HTTP_200_OK) else: return Response("请填写修改意见", status=status.HTTP_400_BAD_REQUEST) return Response("记录状态错误", status=status.HTTP_400_BAD_REQUEST) @action(methods=["put"], detail=True, perms_map={"put": "record_confirm"}) def confirm(self, request, *args, **kwargs): """ 确认 """ obj = self.get_object() if obj.state in ["已报送", "待报送"]: obj.state = "已确认" obj.save() return Response(status=status.HTTP_200_OK) return Response("记录状态错误", status=status.HTTP_400_BAD_REQUEST) class QueryRecordListViewSet(ListModelMixin, GenericViewSet): perms_map = {"get": "queryrecord_view"} queryset = QueryRecord.objects.all() serializer_class = QueryRecordSerializer search_fields = ["user__name", "user__dept__name"] ordering = ["-create_time"] class CMAViewSet(RecordMixin, ModelViewSet): """ CMA检测能力：增删改查 """ perms_map = {"get": "cma_view", "post": "cma_create", "put": "cma_update", "delete": "cma_delete"} queryset = CMA.objects.all() serializer_class = CMASerializer search_fields = ["bzbh", "bzmc", "sszx", "xmmc", "glzz", "dlmc"] filterset_fields = ["sszx", "type", "glzz"] ordering_fields = ["sszx", "dlxh", "lbxh", "xmxh"] ordering = ["create_time", "xmxh"] @action(methods=["get"], detail=False, url_name="cma_group_by", perms_map={"*": "*"}) def group(self, request, pk=None): """ 聚合查询列 """ queryset = self.filter_queryset(self.get_queryset()) ret = [] if request.query_params.get("group_by", None): group_by = request.query_params.get("group_by") group_by_data = list(queryset.values(group_by).annotate(count=Count(group_by)).order_by(group_by)) for i in group_by_data: if i[group_by] and i["count"]: ret.append({"text": i[group_by] + "(" + str(i["count"]) + ")", "value": i[group_by]}) return Response(ret) @action(methods=["post"], detail=False, url_path="deletes", url_name="cma_deletes", perms_map={"post": "cma_deletes"}) def deletes(self, request): array = request.data.get("ids", []) sszx = request.data.get("sszx", None) if array: CMA.objects.filter(pk__in=array).delete() elif sszx: CMA.objects.filter(sszx=sszx).delete() return Response(status=status.HTTP_200_OK) @action(methods=["post"], detail=False, url_path="import", url_name="cma_import", perms_map={"post": "cma_import"}) def cma_import(self, request, pk=None): """ 导入能力 """ filepath = request.data["path"] fullpath = settings.BASE_DIR + filepath import os if fullpath.endswith(".rar"): rar = rarfile.RarFile(fullpath) fulldir = fullpath.replace(".rar", "") os.mkdir(fulldir) os.chdir(fulldir) rar.extractall() rar.close() CMA.objects.filter(type="center").delete() for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xlsx"): import_cma(f, os.path.join(root, f)) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) elif fullpath.endswith(".zip"): fulldir = fullpath.replace(".zip", "") if os.path.exists(fulldir): import shutil shutil.rmtree(fulldir) # 先删除该文件夹 os.mkdir(fulldir) os.chdir(fulldir) CMA.objects.filter(type="center").delete() with zipfile.ZipFile(fullpath, "r") as zzz: zzz.extractall(fulldir) for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xlsx"): import_cma(f.encode("cp437").decode("gbk"), os.path.join(root, f)) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) return Response(status=status.HTTP_200_OK) @action(methods=["post"], detail=False, url_path="import2", url_name="cma_import2", perms_map={"post": "cma_import2"}) def cma_import2(self, request, pk=None): """ 导入能力2 """ filepath = request.data["path"] fullpath = settings.BASE_DIR + filepath import os if fullpath.endswith(".rar"): rar = rarfile.RarFile(fullpath) fulldir = fullpath.replace(".rar", "") os.mkdir(fulldir) os.chdir(fulldir) rar.extractall() rar.close() # CMA.objects.filter(type='sub').delete() for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xlsx"): import_cma2(f, os.path.join(root, f)) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) elif fullpath.endswith(".zip"): fulldir = fullpath.replace(".zip", "") if os.path.exists(fulldir): import shutil shutil.rmtree(fulldir) # 先删除该文件夹 os.mkdir(fulldir) os.chdir(fulldir) # CMA.objects.filter(type='sub').delete() with zipfile.ZipFile(fullpath, "r") as zzz: zzz.extractall(fulldir) for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xlsx"): import_cma2(f.encode("cp437").decode("gbk"), os.path.join(root, f)) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) return Response(status=status.HTTP_200_OK) from django.db.models.query import QuerySet class QualificationViewSet(RecordMixin, ModelViewSet): """ 资质能力：增删改查 """ perms_map = {"get": "qualification_view", "post": "qualificaiton_create", "put": "qualification_update", "delete": "qualification_delete"} queryset = Qualification.objects.all() serializer_class = QualificationSerializer search_fields = ["cma", "cnas", "sszx", "other", "service"] ordering_fields = ["sszx"] ordering = ["sszx"] filterset_fields = ["sszx"] def get_queryset(self): assert self.queryset is not None, "'%s' should either include a `queryset` attribute, " "or override the `get_queryset()` method." % self.__class__.__name__ queryset = self.queryset if isinstance(queryset, QuerySet): # Ensure queryset is re-evaluated on each request. queryset = queryset.all() if hasattr(self.get_serializer_class(), "setup_eager_loading"): queryset = self.get_serializer_class().setup_eager_loading(queryset) return queryset @action(methods=["get"], detail=False, url_name="qualification_group_by", perms_map={"*": "*"}) def group(self, request, pk=None): """ 聚合查询列 """ queryset = self.filter_queryset(self.get_queryset()) ret = [] if request.query_params.get("group_by", None): group_by = request.query_params.get("group_by") group_by_data = list(queryset.values(group_by).annotate(count=Count(group_by)).order_by(group_by)) for i in group_by_data: if i[group_by] and i["count"]: ret.append({"text": i[group_by] + "(" + str(i["count"]) + ")", "value": i[group_by]}) return Response(ret) @action(methods=["post"], detail=False, url_path="import", url_name="qualification_import", perms_map={"post": "qualification_import"}) def qualification_import(self, request, pk=None): filepath = request.data["path"] fullpath = settings.BASE_DIR + filepath if fullpath.endswith(".xlsx"): import_qualification(fullpath) return Response(status=status.HTTP_200_OK) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) class QualificationotherViewSet(RecordMixin, PageOrNot, ModelViewSet): """ 资质能力：增删改查 """ perms_map = {"get": "qualification_view", "post": "qualificationother_create", "put": "qualificationother_update", "delete": "qualificationother_delete"} queryset = Qualificationother.objects.all() serializer_class = QualificationotherSerializer search_fields = ["qualification__cma", "name", "description", "qualification__cnas", "qualification__ssbm__id", "qualification__service"] filterset_fields = ["qualification__ssbm__name"] ordering_fields = ["qualification__ssbm__name"] ordering = ["create_time"] @method_decorator(cache_page(60 * 60 * 2)) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) @action(methods=["post"], detail=False, url_path="deletes", url_name="qualificationother_deletes", perms_map={"post": "qualificationother_deletes"}) def deletes(self, request): array = request.data["ids"] Qualificationother.objects.filter(pk__in=array).delete() return Response(status=status.HTTP_200_OK) @action(methods=["get"], detail=False, url_name="qualification_group_by", perms_map={"*": "*"}) def group(self, request, pk=None): """ 聚合查询列 """ queryset = self.filter_queryset(self.get_queryset()) ret = [] if request.query_params.get("group_by", None): group_by = request.query_params.get("group_by") group_by_data = list(queryset.values(group_by).annotate(count=Count(group_by)).order_by("qualification__ssbm__sort", group_by)) for i in group_by_data: if i[group_by] and i["count"]: ret.append({"text": i[group_by] + "(" + str(i["count"]) + ")", "value": i[group_by]}) return Response(ret) class InspectionViewSet(RecordMixin, ModelViewSet): """ 检验能力：增删改查 """ perms_map = {"get": "inspection_view", "post": "inspection_create", "put": "inspection_update", "delete": "inspection_delete"} queryset = Inspection.objects.all() serializer_class = InspectionSerializer search_fields = ["dlmc", "jydx", "dxxh", "jyxmmc", "jybz", "sszx", "sm"] ordering_fields = ["dlmc"] ordering = ["create_time", "jyxmxh"] filterset_fields = ["sszx"] @action(methods=["post"], detail=False, url_path="deletes", url_name="inspection_deletes", perms_map={"post": "inspection_deletes"}) def deletes(self, request): array = request.data["ids"] Inspection.objects.filter(pk__in=array).delete() return Response(status=status.HTTP_200_OK) @action(methods=["get"], detail=False, url_name="inspection_group_by", perms_map={"*": "*"}) def group(self, request, pk=None): """ 聚合查询列 """ queryset = self.filter_queryset(self.get_queryset()) ret = [] if request.query_params.get("group_by", None): group_by = request.query_params.get("group_by") group_by_data = list(queryset.values(group_by).annotate(count=Count(group_by)).order_by(group_by)) for i in group_by_data: if i[group_by] and i["count"]: ret.append({"text": i[group_by] + "(" + str(i["count"]) + ")", "value": i[group_by]}) return Response(ret) @action(methods=["post"], detail=False, url_path="inspection2", url_name="inspection_inspection2", perms_map={"post": "inspection_inspection2"}) def inspection_import2(self, request, pk=None): """ 导入能力2 """ filepath = request.data["path"] fullpath = settings.BASE_DIR + filepath import os if fullpath.endswith(".rar"): rar = rarfile.RarFile(fullpath) fulldir = fullpath.replace(".rar", "") os.mkdir(fulldir) os.chdir(fulldir) rar.extractall() rar.close() # CMA.objects.filter(type='sub').delete() for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xlsx"): import_inspection(f, os.path.join(root, f)) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) elif fullpath.endswith(".zip"): fulldir = fullpath.replace(".zip", "") if os.path.exists(fulldir): import shutil shutil.rmtree(fulldir) # 先删除该文件夹 os.mkdir(fulldir) os.chdir(fulldir) # CMA.objects.filter(type='sub').delete() with zipfile.ZipFile(fullpath, "r") as zzz: zzz.extractall(fulldir) for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xlsx"): import_inspection(f.encode("cp437").decode("gbk"), os.path.join(root, f)) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) return Response(status=status.HTTP_200_OK) class CNASViewSet(RecordMixin, ModelViewSet): """ CNAS检测能力：增删改查 """ perms_map = {"get": "cma_view", "post": "cnas_create", "put": "cnas_update", "delete": "cnas_delete"} queryset = CNAS.objects.all() serializer_class = CNASSerializer search_fields = ["bzbh", "bzmc", "sszx", "xmmc", "bztk"] ordering_fields = ["bzmc"] filterset_fields = ["sszx"] ordering = ["create_time"] @action(methods=["get"], detail=False, url_name="cnas_group_by", perms_map={"*": "*"}) def group(self, request, pk=None): """ 聚合查询列 """ queryset = self.filter_queryset(self.get_queryset()) ret = [] if request.query_params.get("group_by", None): group_by = request.query_params.get("group_by") group_by_data = list(queryset.values(group_by).annotate(count=Count(group_by)).order_by(group_by)) for i in group_by_data: if i[group_by] and i["count"]: ret.append({"text": i[group_by] + "(" + str(i["count"]) + ")", "value": i[group_by]}) return Response(ret) @action(methods=["post"], detail=False, url_path="import", url_name="cnas_import", perms_map={"post": "cnas_import"}) def cnas_import(self, request, pk=None): """ 导入能力 """ filepath = request.data["path"] fullpath = settings.BASE_DIR + filepath import os if fullpath.endswith(".rar"): rar = rarfile.RarFile(fullpath) fulldir = fullpath.replace(".rar", "") os.mkdir(fulldir) os.chdir(fulldir) rar.extractall() rar.close() CNAS.objects.all().delete() for root, dirs, files in os.walk(fulldir): for f in files: if f.endswith(".xls"): return Response("不支持旧xls格式", status=status.HTTP_400_BAD_REQUEST) import_cnas(f, os.path.join(root, f)) elif fullpath.endswith(".zip"): fulldir = fullpath.replace(".zip", "") if os.path.exists(fulldir): import shutil shutil.rmtree(fulldir) # 先删除该文件夹 os.mkdir(fulldir) os.chdir(fulldir) CNAS.objects.all().delete() with zipfile.ZipFile(fullpath, "r") as zzz: zzz.extractall(fulldir) for root, dirs, files in os.walk(fulldir): for f in files: import_cnas(f.encode("cp437").decode("gbk"), os.path.join(root, f)) return Response(status=status.HTTP_200_OK) class CorrectViewSet(RecordMixin, ModelViewSet): """ 校验能力：增删改查 """ perms_map = {"get": "correct_view", "post": "correct_create", "put": "correct_update", "delete": "correct_delete"} queryset = Correct.objects.all() serializer_class = CorrectSerializer search_fields = ["dlmc", "lbmc", "bclmc", "jzgc"] ordering_fields = ["dlmc"] ordering = ["create_time", "dlmc"] filterset_fields = ["ssbm", "ssgs"] @action(methods=["post"], detail=False, perms_map={"post": "correct_delete"}) def deletes(self, request): array = request.data["ids"] Correct.objects.filter(pk__in=array).delete() return Response(status=status.HTTP_200_OK) @action(methods=["get"], detail=False, perms_map={"*": "*"}) def group(self, request, pk=None): """ 聚合查询列 """ queryset = self.filter_queryset(self.get_queryset()) ret = [] if request.query_params.get("group_by", None): group_by = request.query_params.get("group_by") group_by_data = list(queryset.values(group_by).annotate(count=Count(group_by)).order_by(group_by)) for i in group_by_data: if i[group_by] and i["count"]: ret.append({"text": i[group_by] + "(" + str(i["count"]) + ")", "value": i[group_by]}) return Response(ret) @action(methods=["post"], detail=False, url_path="import") def correct_import(self, request, pk=None): """ 导入校验能力 """ filepath = request.data["path"] fullpath = settings.BASE_DIR + filepath if fullpath.endswith(".xlsx"): ret = import_correct(fullpath) if ret[0]: return Response() return Response(ret[1], status=status.HTTP_400_BAD_REQUEST) else: return Response("不支持非xlsx格式", status=status.HTTP_400_BAD_REQUEST) def import_qualification(path): wb = load_workbook(path) sheet = wb.worksheets[0] i = 3 max_row = sheet.max_row obj = {} Qualificationother.objects.all().delete() Qualification.objects.all().delete() while i < max_row + 1: sszx = sheet["b" + str(i)].value cma = sheet["c" + str(i)].value cnas = sheet["d" + str(i)].value service = sheet["g" + str(i)].value name = sheet["e" + str(i)].value description = sheet["f" + str(i)].value if sszx: if Organization.objects.filter(name=sszx).exists(): ssbm = Organization.objects.get(name=sszx) if Qualification.objects.filter(ssbm=ssbm).exists(): qualification = Qualification.objects.get(ssbm=ssbm) Qualificationother.objects.create(qualification=qualification, name=name, description=description) else: qualification = Qualification.objects.create(ssbm=ssbm) qualification.cma = cma qualification.cnas = cnas qualification.service = service qualification.save() obj = qualification Qualificationother.objects.create(qualification=qualification, name=name, description=description) else: Qualificationother.objects.create(qualification=obj, name=name, description=description) i = i + 1 def import_cma(filename, path): wb = load_workbook(path) sheet = wb.worksheets[0] datalist = [] sszx = filename.replace(".xlsx", "").replace("14检验检测能力申请表-", "") i = 4 while sheet["b" + str(i)].value: data = {} data["dlxh"] = sheet["a" + str(i)].value data["dlmc"] = sheet["b" + str(i)].value data["lbxh"] = sheet["c" + str(i)].value data["lbmc"] = sheet["d" + str(i)].value data["xmxh"] = sheet["e" + str(i)].value data["xmmc"] = sheet["f" + str(i)].value data["bzmc"] = sheet["g" + str(i)].value data["bzbh"] = sheet["h" + str(i)].value data["xzfw"] = sheet["i" + str(i)].value data["bz"] = sheet["j" + str(i)].value data["sszx"] = sszx data["type"] = "center" datalist.append(CMA(**data)) i = i + 1 CMA.objects.bulk_create(datalist) import xlrd def import_cnas(filename, path): sheet = xlrd.open_workbook(path).sheet_by_name("检测能力范围") datalist = [] sszx = filename.replace(".xlsx", "").replace(".xls", "").replace("检测能力范围(含能源之星)-", "") i = 2 lbmc = "" xmmc = "" while i < sheet.nrows: data = {} data_list = sheet.row_values(i) if data_list[1]: lbmc = data_list[1] if data_list[6]: xmmc = data_list[6] data["lbmc"] = lbmc data["xmmc"] = xmmc data["bzmc"] = data_list[11] data["bzbh"] = data_list[13] data["bztk"] = data_list[15] data["sszx"] = sszx data["sm"] = data_list[17] datalist.append(CNAS(**data)) i = i + 1 CNAS.objects.bulk_create(datalist) def import_cma2(filename, path): wb = load_workbook(path, data_only=True) sheet = wb.worksheets[0] datalist = [] sszx = filename.split("-")[0] # if CMA.objects.filter(sszx=sszx, type='sub').exists(): # CMA.objects.filter(sszx=sszx, type='sub').delete() i = 3 max_row = sheet.max_row defaultv = {} while i < max_row + 1: data = {} if sheet["a" + str(i)].value: data["dlxh"] = sheet["a" + str(i)].value defaultv["dlxh"] = data["dlxh"] else: data["dlxh"] = defaultv["dlxh"] if sheet["b" + str(i)].value: data["dlmc"] = sheet["b" + str(i)].value defaultv["dlmc"] = data["dlmc"] else: data["dlmc"] = defaultv["dlmc"] if sheet["c" + str(i)].value: data["lbxh"] = sheet["c" + str(i)].value defaultv["lbxh"] = data["lbxh"] else: data["lbxh"] = defaultv.get("lbxh", "") if sheet["d" + str(i)].value: data["lbmc"] = sheet["d" + str(i)].value defaultv["lbmc"] = data["lbmc"] else: data["lbmc"] = defaultv["lbmc"] if sheet["e" + str(i)].value: data["xmxh"] = sheet["e" + str(i)].value defaultv["xmxh"] = data["xmxh"] else: data["xmxh"] = defaultv.get("xmxh", "") if sheet["f" + str(i)].value: data["xmmc"] = sheet["f" + str(i)].value defaultv["xmmc"] = data["xmmc"] else: data["xmmc"] = defaultv["xmmc"] if sheet["g" + str(i)].value: data["bzmc"] = sheet["g" + str(i)].value defaultv["bzmc"] = data["bzmc"] else: data["bzmc"] = defaultv["bzmc"] if sheet["h" + str(i)].value: data["bzbh"] = sheet["h" + str(i)].value defaultv["bzbh"] = data["bzbh"] else: data["bzbh"] = defaultv["bzbh"] data["xzfw"] = sheet["i" + str(i)].value if (sheet["i" + str(i)].value and sheet["i" + str(i)].value != "无") else None data["bz"] = sheet["j" + str(i)].value if (sheet["j" + str(i)].value and sheet["j" + str(i)].value != "无") else None data["glzz"] = sheet["k" + str(i)].value if (sheet["k" + str(i)].value and sheet["k" + str(i)].value != "无") else None data["sszx"] = sszx data["type"] = "sub" datalist.append(CMA(**data)) i = i + 1 CMA.objects.bulk_create(datalist) def import_inspection(filename, path): wb = load_workbook(path, data_only=True) sheet = wb.worksheets[0] datalist = [] sszx = filename.split("-")[0] if Inspection.objects.filter(sszx=sszx).exists(): Inspection.objects.filter(sszx=sszx).delete() i = 3 max_row = sheet.max_row defaultv = {} while i < max_row + 1: data = {} if sheet["a" + str(i)].value: data["dlxh"] = sheet["a" + str(i)].value defaultv["dlxh"] = data["dlxh"] else: data["dlxh"] = defaultv["dlxh"] if sheet["b" + str(i)].value: data["dlmc"] = sheet["b" + str(i)].value defaultv["dlmc"] = data["dlmc"] else: data["dlmc"] = defaultv["dlmc"] if sheet["c" + str(i)].value: data["dxxh"] = sheet["c" + str(i)].value defaultv["dxxh"] = data["dxxh"] else: data["dxxh"] = defaultv["dxxh"] if sheet["d" + str(i)].value: data["jydx"] = sheet["d" + str(i)].value defaultv["jydx"] = data["jydx"] else: data["jydx"] = defaultv["jydx"] if sheet["e" + str(i)].value: data["jyxmxh"] = sheet["e" + str(i)].value elif sheet["e3"].value: # 该表存在项目序号 m = i - 1 while True: if sheet["e" + str(m)].value: data["jyxmxh"] = sheet["e" + str(m)].value break m = m - 1 else: # 该表没有项目序号，自己定 pass if sheet["f" + str(i)].value: data["jyxmmc"] = sheet["f" + str(i)].value else: m = i - 1 while True: if sheet["f" + str(m)].value: data["jyxmmc"] = sheet["f" + str(m)].value break m = m - 1 if sheet["g" + str(i)].value: data["jybz"] = sheet["g" + str(i)].value elif sheet["g3"].value: m = i - 1 while True: if sheet["g" + str(m)].value: data["jybz"] = sheet["g" + str(m)].value break m = m - 1 if sheet["h" + str(i)].value: data["sm"] = sheet["h" + str(i)].value elif sheet["h3"].value: m = i - 1 while True: if sheet["h" + str(m)].value: data["sm"] = sheet["h" + str(m)].value break m = m - 1 if sheet["i" + str(i)].value: data["sxrq"] = sheet["i" + str(i)].value elif sheet["i3"].value: m = i - 1 while True: if sheet["i" + str(m)].value: data["sxrq"] = sheet["i" + str(m)].value break m = m - 1 data["sszx"] = sszx datalist.append(Inspection(**data)) i = i + 1 Inspection.objects.bulk_create(datalist) def import_correct(path): wb = load_workbook(path, data_only=True) sheet = wb.worksheets[0] datalist = [] ssgs = sheet["k3"].value try: ssbm = Organization.objects.get(name=ssgs) except: return False, ssgs + "不存在" if Correct.objects.filter(ssbm=ssbm).exists(): Correct.objects.filter(ssbm=ssbm).delete() i = 3 max_row = sheet.max_row defaultv = {} while i < max_row + 1: data = {} if sheet["a" + str(i)].value: data["dlxh"] = sheet["a" + str(i)].value defaultv["dlxh"] = data["dlxh"] else: data["dlxh"] = defaultv["dlxh"] if sheet["b" + str(i)].value: data["dlmc"] = sheet["b" + str(i)].value defaultv["dlmc"] = data["dlmc"] else: data["dlmc"] = defaultv["dlmc"] if sheet["c" + str(i)].value: data["lbxh"] = sheet["c" + str(i)].value defaultv["lbxh"] = data["lbxh"] else: data["lbxh"] = defaultv["lbxh"] if sheet["d" + str(i)].value: data["lbmc"] = sheet["d" + str(i)].value defaultv["lbmc"] = data["lbmc"] else: data["lbmc"] = defaultv["lbmc"] if sheet["e" + str(i)].value: data["bclxh"] = sheet["e" + str(i)].value defaultv["bclxh"] = data["bclxh"] else: data["bclxh"] = defaultv["bclxh"] if "bclxh" in defaultv else None if sheet["f" + str(i)].value: data["bclmc"] = sheet["f" + str(i)].value defaultv["bclmc"] = data["bclmc"] else: data["bclmc"] = defaultv["bclmc"] if sheet["g" + str(i)].value: data["jzgc"] = sheet["g" + str(i)].value defaultv["jzgc"] = data["jzgc"] else: data["jzgc"] = defaultv["jzgc"] if sheet["h" + str(i)].value: data["clfw"] = sheet["h" + str(i)].value defaultv["clfw"] = data["clfw"] else: data["clfw"] = defaultv["clfw"] data["zqddj"] = sheet["i" + str(i)].value if (sheet["i" + str(i)].value and sheet["i" + str(i)].value != "无") else None data["note"] = sheet["j" + str(i)].value if (sheet["j" + str(i)].value and sheet["j" + str(i)].value != "无") else None data["ssgs"] = ssgs data["ssbm"] = ssbm datalist.append(Correct(**data)) i = i + 1 Correct.objects.bulk_create(datalist) return True, "" from django.db.models.functions import Cast def merge_cnas(request): for i in CNAS.objects.all(): objs = Ability.objects.filter(lbmc=i.lbmc, xmmc=i.xmmc, bzbh=i.bzbh, bztk=i.bztk, bzmc=i.bzmc) if objs.exists(): obj = objs[0] obj.cnas_ok = True if obj.cnas: obj.cnas = obj.cnas + "," + i.sszx obj.save() else: obj.cnas = i.sszx obj.save() print("已修改--" + obj.xmmc + "-" + obj.cnas) else: obj = Ability() obj.lbmc = i.lbmc obj.xmmc = i.xmmc obj.bzmc = i.bzmc obj.bzbh = i.bzbh obj.bztk = i.bztk obj.cnas = i.sszx obj.cnas_ok = True obj.save() print("已添加--" + obj.xmmc + obj.bzmc) return Response({}) def merge_cma(request): for i in CMA.objects.filter(type="center"): bztk = i.bzbh.split(" ")[-1] bzbh = i.bzbh.rstrip(bztk).strip() xmxh = i.xmxh.strip() if len(bzbh) < 8: bzbh = i.bzbh bztk = "" objs = Ability.objects.filter(lbmc=i.lbmc, xmmc=i.xmmc, bzbh=bzbh, bztk=bztk, xmxh=xmxh, bzmc=i.bzmc) if objs.exists(): obj = objs[0] obj.cma = obj.cma + "," + i.sszx obj.save() print("已修改--" + obj.xmmc + "-" + obj.cma) else: obj = Ability() obj.dlxh = i.dlxh obj.dlmc = i.dlmc obj.lbxh = i.lbxh obj.lbmc = i.lbmc obj.xmxh = i.xmxh obj.xmmc = i.xmmc obj.bzmc = i.bzmc obj.bzbh = bzbh obj.bztk = bztk obj.xzfw = i.xzfw obj.bz = i.bz obj.cma = i.sszx obj.cma_ok = True obj.save() print("已添加--" + obj.xmmc + obj.bzmc) return Response({}) def correct_ability(request): zxdict = {"测试中心": "B", "玻璃中心": "E", "腐蚀中心": "P", "光伏中心": "L", "耐火中心": "Q", "石材中心": "R", "水泥中心": "D", "质检中心": "C"} placedict = { "玻璃中心-管庄": "E", "玻璃中心-密云": "E", "玻璃中心-永顺": "E", "测试中心-管庄": "B", "测试中心-光华路": "B", "测试中心-国检三层": "B", "测试中心-密云": "B", "测试中心-顺义种植园": "B", "测试中心-宋庄": "B", "测试中心-通州种植园": "B", "测试中心-永顺": "B", "腐蚀中心": "P", "光伏中心-管庄": "L", "光伏中心-密云": "L", "耐火中心": "Q", "石材中心": "R", "水泥中心": "D", "质检中心-管庄": "C", "质检中心-光华路": "C", "质检中心-国检三层": "c", "质检中心-密云": "C", "质检中心-顺义种植园": "C", "质检中心-宋庄": "C", "质检中心-通州种植园": "C", "质检中心-永顺": "C", } for i in Ability.objects.all(): # i.cma_o = i.cma i.cma = i.cma_o i.cma_oplace = i.cma_o for key in zxdict: if key in i.cma_o: i.cma = i.cma.replace(key, zxdict[key]) for key in placedict: if key in i.cma_o: i.cma_oplace = i.cma_oplace.replace(key, placedict[key]) i.save() print("已修改" + i.cma) def correct_cma(request): for i in CMA.objects.all(): i.sszx = i.sszx.replace("14检验检测能力申请表-", "") i.save() return Response()