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 = ( list(category.translations.values("lang_code", "name")) or [] ) # 子表字典项 items = category.items.order_by("sort", "id") item_list = [] for item in items: item_i18n_name = ( list(item.translations.values("lang_code", "name")) 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, "itemNames": 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, "names": i18n_name, "items": item_list }) return response.json(0, { "list": result, "total": total }) except Exception as e: return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(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, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(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, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(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: 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() if names: # 删除原有多语言并重新插入 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 ) if not items: return response.json(0) # 删除原有子项及多语言 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, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(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, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))