数据字典后台管理

AdminController/DataDictManageController.py

@@ -0,0 +1,357 @@
+import json
+import time
+
+from django.db import transaction
+from django.db.models import Q
+from django.views import View
+
+from Model.models import DictCategory, DictCategoryI18n, DictItemI18n, DictItem
+from Object.ResponseObject import ResponseObject
+from Object.TokenObject import TokenObject
+
+
+class DataDictView(View):
+    def get(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        request_dict = request.GET
+        return self.validation(request_dict, request, operation)
+
+    def post(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        request_dict = request.POST
+        return self.validation(request_dict, request, operation)
+
+    def validation(self, request_dict, request, operation):
+        language = request_dict.get('language', 'en')
+        response = ResponseObject(language, 'pc')
+        tko = TokenObject(
+            request.META.get('HTTP_AUTHORIZATION'),
+            returntpye='pc')
+        response.lang = tko.lang
+        if operation == "getDictCategory":
+            return self.get_dict_category(request_dict, response)
+        elif operation == "getDictCategoryList":
+            return self.dict_categories_list(request_dict, response)
+        if tko.code != 0:
+            return response.json(tko.code)
+        else:
+            response.lang = tko.lang
+            userID = tko.userID
+            if operation == 'addDictCategory':
+                return self.add_dict_category(userID, request_dict, response)
+            elif operation == 'editDictCategory':
+                return self.edit_dict_category(userID, request_dict, response)
+            elif operation == 'deleteDictCategory':
+                return self.delete_dict_category(request_dict, response)
+            else:
+                return response.json(414)
+
+    @staticmethod
+    def dict_categories_list(request_dict, response):
+        code = request_dict.get("typeCode", None)
+        status = request_dict.get("status", None)
+        scenes = request_dict.get("scenes", None)
+        page = int(request_dict.get("page", 1))
+        page_size = int(request_dict.get("pageSize", 10))
+        try:
+            query = Q()
+            if code:
+                query &= Q(code__icontains=code)
+            if status:
+                query &= Q(status=bool(int(status)))
+            if scenes:
+                query &= Q(scenes=int(scenes))
+
+            category_qs = DictCategory.objects.filter(query).order_by("-id")
+
+            total = category_qs.count()
+            start = (page - 1) * page_size
+            end = start + page_size
+            categories = category_qs[start:end]
+
+            result = []
+            for category in categories:
+                # 类别名称国际化
+                i18n_name = (
+                        category.translations.values_list("name", flat=True).first()
+                        or ""
+                )
+
+                # 子表字典项
+                items = category.items.filter(status=True).order_by("sort", "id")
+                item_list = []
+                for item in items:
+                    item_i18n_name = (
+                            item.translations.values_list("name", flat=True).first()
+                            or ""
+                    )
+                    item_list.append({
+                        "id": item.id,
+                        "code": item.code,
+                        "sort": item.sort,
+                        "status": item.status,
+                        "remark": item.remark,
+                        "createdTime": item.created_time,
+                        "createdBy": item.created_by,
+                        "updatedTime": item.updated_time,
+                        "updatedBy": item.updated_by,
+                        "name": item_i18n_name
+                    })
+
+                result.append({
+                    "id": category.id,
+                    "code": category.code,
+                    "scenes": category.scenes,
+                    "status": category.status,
+                    "remark": category.remark,
+                    "extData": category.ext_data,
+                    "createdTime": category.created_time,
+                    "createdBy": category.created_by,
+                    "updatedTime": category.updated_time,
+                    "updatedBy": category.updated_by,
+                    "name": i18n_name,
+                    "items": item_list
+                })
+
+            return response.json(0, {
+                "list": result,
+                "total": total
+            })
+
+        except Exception as e:
+            return response.json(500, msg=f"查询失败: {str(e)}")
+
+    @staticmethod
+    def get_dict_category(request_dict, response):
+        type_codes = request_dict.get("typeCodes", None)
+        lang_code = request_dict.get("langCode", "cn")
+        scenes = request_dict.get("scenes", 1)
+        try:
+            if not type_codes:
+                return response.json(444)
+            type_code_list = type_codes.split(",")
+            category_qs = DictCategory.objects.filter(code__in=type_code_list, status=True, scenes=scenes)
+
+            if not category_qs.exists():
+                return response.json(0)
+
+            result_list = []
+            for category in category_qs:
+                # 获取类别名称（国际化）
+                category_name = (
+                        category.translations.filter(lang_code=lang_code).values_list("name", flat=True).first()
+                        or ""
+                )
+
+                # 构建返回结构
+                result = {
+                    "typeCode": category.code,
+                    "scenes": category.scenes,
+                    "status": category.status,
+                    "remark": category.remark,
+                    "name": category_name,
+                    "items": []
+                }
+
+                # 获取字典项及其多语言名称
+                items = category.items.filter(status=True).order_by("sort", "id")
+                for item in items:
+                    item_name = (
+                            item.translations.filter(lang_code=lang_code).values_list("name", flat=True).first()
+                            or ""
+                    )
+                    result["items"].append({
+                        "code": item.code,
+                        "name": item_name,
+                        "remark": item.remark,
+                        "sort": item.sort
+                    })
+
+                result_list.append(result)
+            return response.json(0, result_list)
+
+        except Exception as e:
+            return response.json(500, msg=f"查询失败: {str(e)}")
+
+
+    @staticmethod
+    def add_dict_category(user_id, request_dict, response):
+        type_code = request_dict.get('typeCode', None)
+        scenes = request_dict.get("scenes", 1)
+        status = request_dict.get("status", 1)
+        remark = request_dict.get("remark", "")
+        names = request_dict.get("names", [])
+        items = request_dict.get("items", [])
+        now_time = int(time.time())
+
+        try:
+            if not type_code or not names:
+                return response.json(444)
+
+            names = json.loads(names)
+            items = json.loads(items)
+            if DictCategory.objects.filter(code=type_code).exists():
+                return response.json(174)
+
+            with transaction.atomic():
+                # 创建类别   数据
+                category = DictCategory.objects.create(
+                    code=type_code,
+                    scenes=scenes,
+                    status=status,
+                    remark=remark,
+                    created_time=now_time,
+                    updated_time=now_time,
+                    created_by=user_id,
+                    updated_by=user_id
+                )
+
+                # 添加多语言名称
+                for name_entry in names:
+                    DictCategoryI18n.objects.create(
+                        category=category,
+                        lang_code=name_entry["lang"],
+                        name=name_entry["text"],
+                        created_time=now_time,
+                        updated_time=now_time,
+                        created_by=user_id,
+                        updated_by=user_id
+                    )
+
+                # 创建选项表数据
+                for item in items:
+                    item_obj = DictItem.objects.create(
+                        category=category,
+                        code=item["code"],
+                        sort=item.get("sort", 0),
+                        status=item.get("status", True),
+                        remark=item.get("remark", ""),
+                        created_time=now_time,
+                        updated_time=now_time,
+                        created_by=user_id,
+                        updated_by=user_id
+                    )
+
+                    for name_entry in item.get("names", []):
+                        DictItemI18n.objects.create(
+                            item=item_obj,
+                            lang_code=name_entry["lang"],
+                            name=name_entry["text"],
+                            created_time=now_time,
+                            updated_time=now_time,
+                            created_by=user_id,
+                            updated_by=user_id
+                        )
+
+            return response.json(0)
+        except Exception as e:
+            return response.json(500, f'添加失败: {str(e)}')
+
+
+    @staticmethod
+    def edit_dict_category(user_id, request_dict, response):
+        category_id = request_dict.get("categoryId", None)
+        scenes = request_dict.get("scenes", 1)
+        status = request_dict.get("status", 1)
+        remark = request_dict.get("remark", "")
+        names = request_dict.get("names", [])
+        items = request_dict.get("items", [])
+        now_time = int(time.time())
+
+        try:
+            if not category_id or not names:
+                return response.json(444)
+
+            names = json.loads(names)
+            items = json.loads(items)
+
+            category = DictCategory.objects.filter(id=category_id).first()
+            if not category:
+                return response.json(404, msg="字典类别不存在")
+
+            with transaction.atomic():
+                # 更新主表字段（不允许改 typeCode）
+                category.status = status
+                category.scenes = scenes
+                category.remark = remark
+                category.updated_time = now_time
+                category.updated_by = user_id
+                category.save()
+
+                # 删除原有多语言并重新插入
+                DictCategoryI18n.objects.filter(category=category).delete()
+                for name_entry in names:
+                    DictCategoryI18n.objects.create(
+                        category=category,
+                        lang_code=name_entry["lang"],
+                        name=name_entry["text"],
+                        created_time=now_time,
+                        updated_time=now_time,
+                        created_by=user_id,
+                        updated_by=user_id
+                    )
+
+                # 删除原有子项及多语言
+                old_items = DictItem.objects.filter(category=category)
+                DictItemI18n.objects.filter(item__in=old_items).delete()
+                old_items.delete()
+
+                # 插入新子项及其国际化
+                for item in items:
+                    item_obj = DictItem.objects.create(
+                        category=category,
+                        code=item["code"],
+                        sort=item.get("sort", 0),
+                        status=item.get("status", True),
+                        remark=item.get("remark", ""),
+                        created_time=now_time,
+                        updated_time=now_time,
+                        created_by=user_id,
+                        updated_by=user_id
+                    )
+
+                    for name_entry in item.get("names", []):
+                        DictItemI18n.objects.create(
+                            item=item_obj,
+                            lang_code=name_entry["lang"],
+                            name=name_entry["text"],
+                            created_time=now_time,
+                            updated_time=now_time,
+                            created_by=user_id,
+                            updated_by=user_id
+                        )
+
+            return response.json(0)
+
+        except Exception as e:
+            return response.json(500, f"编辑失败: {str(e)}")
+
+
+    @staticmethod
+    def delete_dict_category(request_dict, response):
+        category_id = request_dict.get("categoryId", None)
+
+        try:
+            if not category_id:
+                return response.json(444)
+
+            category = DictCategory.objects.filter(id=category_id).first()
+            if not category:
+                return response.json(173)
+
+            with transaction.atomic():
+                # 删除子项和对应的国际化
+                items = DictItem.objects.filter(category=category)
+                DictItemI18n.objects.filter(item__in=items).delete()
+                items.delete()
+
+                # 删除类别国际化和主表记录
+                DictCategoryI18n.objects.filter(category=category).delete()
+                category.delete()
+
+            return response.json(0)
+
+        except Exception as e:
+            return response.json(500, f"删除失败: {str(e)}")

Ansjer/urls.py

@@ -5,7 +5,7 @@ from django.urls import re_path, path
 from AdminController import UserManageController, RoleController, MenuController, TestServeController, \
     ServeManagementController, LogManagementController, DeviceManagementController, VersionManagementController, \
     AiServeController, SurveysManageController, SerialManageController, IcloudManagementController, CampaignController, \
-    MessageMangementController, EvaluationActivityController
+    MessageMangementController, EvaluationActivityController, DataDictManageController
 from AdminController.CloudServiceManage import AgentDeviceController, AgentCustomerController, AgentOrderController
 from Controller import FeedBack, EquipmentOTA, EquipmentInfo, AdminManage, AppInfo, \
     Test, MealManage, DeviceManage,  EquipmentStatus, SysManage, DeviceLog, LogAccess, \
@@ -342,6 +342,8 @@ urlpatterns = [
     re_path('activityManagement/(?P<operation>.*)', EvaluationActivityController.EvaluationActivityView.as_view()),
     # 活动中心
     re_path('activityCenter/(?P<operation>.*)', ActivityCenterController.ActivityCenterView.as_view()),
+    # 数据字典
+    re_path('dataDictView/(?P<operation>.*)', DataDictManageController.DataDictView.as_view()),
     # 后台界面接口 -------------------------------------------------------------------------------------------------------
 
     # 定时任务接口