# +====================================================================== # | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] # +---------------------------------------------------------------------- # | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 # +---------------------------------------------------------------------- # | 开源协议: Apache-2.0 # | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 # +---------------------------------------------------------------------- # | 官方网站: https://www.djangoadmin.cn # +---------------------------------------------------------------------- # | 研发团队: 小蚂蚁云团队 荣誉出品 # +====================================================================== # | 版权和免责声明: # | 一、知识产权声明 # | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 # | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， # | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 # | 二、使用限制与禁止条款 # | 您理解并同意，禁止将本产品用于以下任何行为： # | 1. 任何违反中华人民共和国法律法规的用途； # | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； # | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 # | 三、责任豁免 # | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 # | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 # | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 # | 四、违约与法律责任 # | 如发现任何违反本声明的行为，本团队有权： # | 1. 无条件配合公安机关及其他有权机关的调查取证工作； # | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； # | 3. 保留进一步采取刑事控告等法律措施的权利。 # | 五、许可范围说明 # | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 # | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 # | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 # | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 # | 成的一切经济损失及商誉损害。 # | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 # | 许可协议》附件为准。 # +====================================================================== """ 文章分类管理服务模块 ============================================================= 该模块实现了文章分类管理相关的所有业务逻辑。 包括文章分类的分页查询、详情查询、添加、更新、删除等功能。 包含分类名称, 分类编码, 上级ID等属性 命名规范： - 函数参数使用驼峰命名（如 id, pageNo） - 表单字段使用下划线命名（如 category_name, category_code, parent_id, sort, note）- 与数据库字段保持一致 - 返回数据使用驼峰命名（如 createUser, updateTime）- 与前端规范保持一致 - 私有函数以下划线开头（如 _apply_filters, _build_page_data） 功能模块： 1. 分页查询：支持按分类名称搜索、分类编码搜索、上级ID搜索、分类备注搜索，按sort字段排序 2. 详情查询：根据ID获取单个文章分类的详细信息 3. 数据维护：文章分类的添加、更新、删除（支持批量） 4. 数据列表：获取所有启用的文章分类列表（用于下拉选择等场景） """ import json import logging from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage, EmptyPage from application.category import forms from application.category.models import Category from config.env import DJANGO_FILE_URL from constant.constants import PAGE_SIZE from utils import R, regular from utils.common import get_file_url, format_datetime, parse_request_body, save_file, save_content from utils.security import get_username # ============================================================= # 分页查询服务 # ============================================================= def get_category_page(request): """ 查询文章分类分页数据 Args: request: HttpRequest对象，包含查询参数 - pageNo: 当前页码（可选，默认为1） - pageSize: 每页记录数（可选，默认为PAGE_SIZE） - category_name: 分类名称（可选，模糊查询） - category_code: 分类编码（可选，精确匹配） - parent_id: 上级ID（可选，精确匹配） - note: 分类备注（可选，精确匹配） Returns: R: 包含分页数据的JSON响应，格式为： { "records": 文章分类列表, "total": 总记录数, "size": 每页大小, "current": 当前页码, "pages": 总页数 } """ try: # 获取分页参数 page_no = int(request.GET.get('pageNo', 1)) page_size = int(request.GET.get('pageSize', PAGE_SIZE)) # 构建基础查询（只查询未删除的记录） query = Category.objects.filter(is_delete=False) # 应用筛选条件 query = _apply_filters(query, request) # 按排序字段升序排列 query = query.order_by("sort") # 执行分页查询 paginator = Paginator(query, page_size) try: page_list = paginator.page(page_no) except (PageNotAnInteger, InvalidPage, EmptyPage): # 页码无效时默认返回第一页 page_list = paginator.page(1) page_no = 1 # 构建并返回分页数据 page_data = _build_page_data(page_list, paginator, page_no, page_size) return R.ok(data=page_data) except Exception as e: logging.error(f"查询文章分类分页数据异常: {str(e)}") return R.failed(msg="查询失败，请稍后重试") def _apply_filters(queryset, request): """ 应用查询筛选条件（私有辅助函数） Args: queryset: 初始查询集 request: HttpRequest对象，包含筛选参数 Returns: QuerySet: 应用筛选后的查询集 """ # 分类名称筛选（模糊查询） category_name = request.GET.get('category_name') if category_name: queryset = queryset.filter(category_name__contains=category_name) # 分类编码筛选（精确匹配） category_code = request.GET.get('category_code') if category_code: queryset = queryset.filter(category_code=category_code) # 上级ID筛选（精确匹配） parent_id = request.GET.get('parent_id') if parent_id: queryset = queryset.filter(parent_id=parent_id) # 分类备注筛选（精确匹配） note = request.GET.get('note') if note: queryset = queryset.filter(note=note) return queryset def _build_page_data(page_list, paginator, page_no, page_size): """ 构建分页返回数据（私有辅助函数） Args: page_list: 当前页数据列表（Page对象） paginator: 分页器对象 page_no: 当前页码 page_size: 每页大小 Returns: dict: 标准格式的分页数据 """ records = [] for item in page_list: record = { 'id': item.id, 'categoryName': item.category_name, 'categoryCode': item.category_code, 'parentId': item.parent_id, 'sort': item.sort, 'note': item.note, 'createUser': item.create_user, 'createTime': format_datetime(item.create_time), 'updateUser': item.update_user, 'updateTime': format_datetime(item.update_time), } records.append(record) return { 'records': records, 'total': paginator.count, # 总记录数 'size': page_size, # 每页大小 'current': page_no, # 当前页码 'pages': paginator.num_pages # 总页数 } # ============================================================= # 列表查询服务（不分页） # ============================================================= def get_category_list(request): """ 获取文章分类列表（不分页，支持筛选条件） 与分页查询的区别：返回所有符合条件的记录，不分页。 通常用于导出数据或内部调用。 Args: request: HttpRequest对象，包含查询参数 - category_name: 分类名称（可选，模糊查询） - category_code: 分类编码（可选，精确匹配） - parent_id: 上级ID（可选，精确匹配） - note: 分类备注（可选，精确匹配） Returns: list: 文章分类列表，每个元素为文章分类模型字典 查询失败时返回空列表 """ try: # 构建基础查询 query = Category.objects.filter(is_delete=False) # 应用筛选条件（复用 _apply_filters 方法） query = _apply_filters(query, request) # 按排序字段升序排列 query = query.order_by("sort") # 构建返回数据 records = [] for item in query: record = { 'id': item.id, 'categoryName': item.category_name, 'categoryCode': item.category_code, 'parentId': item.parent_id, 'sort': item.sort, 'note': item.note, 'createUser': item.create_user, 'createTime': format_datetime(item.create_time), 'updateUser': item.update_user, 'updateTime': format_datetime(item.update_time), } records.append(record) return records except Exception as e: logging.error(f"查询文章分类列表异常: {str(e)}") return [] # ============================================================= # 详情查询服务 # ============================================================= def get_category_detail(id): """ 根据ID查询文章分类详情 Args: id: 文章分类ID Returns: dict/None: 文章分类详情字典，不存在时返回None """ try: # 查询未删除的文章分类记录 instance = Category.objects.filter(is_delete=False, id=id).first() if not instance: return None # 返回详情数据 return { 'id': instance.id, 'categoryName': instance.category_name, 'categoryCode': instance.category_code, 'parentId': instance.parent_id, 'sort': instance.sort, 'note': instance.note, } except Exception as e: logging.error(f"查询文章分类详情异常，ID: {id}, 错误: {str(e)}") return None # ============================================================= # 添加服务 # ============================================================= def add_category(request): """ 添加文章分类 处理流程： 1. 解析请求体中的JSON数据 2. 表单验证 3. 创建数据库记录 Args: request: HttpRequest对象，包含文章分类信息的JSON数据 Returns: R: 操作结果（成功/失败） """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) # 表单验证 form = forms.CategoryForm(data) if not form.is_valid(): return R.failed(msg=regular.get_err(form)) try: cleaned_data = form.cleaned_data # 创建数据库记录 Category.objects.create( category_name=cleaned_data.get('categoryName'), category_code=cleaned_data.get('categoryCode'), parent_id=cleaned_data.get('parentId'), sort=cleaned_data.get('sort'), note=cleaned_data.get('note'), create_user=get_username(request), update_user=get_username(request) ) return R.ok(msg="创建成功") except Exception as e: logging.error(f"添加文章分类异常: {str(e)}") return R.failed(msg="添加失败，请稍后重试") # ============================================================= # 更新服务 # ============================================================= def update_category(request): """ 更新文章分类 处理流程： 1. 解析请求体中的JSON数据 2. 验证文章分类ID 3. 表单验证 4. 查询现有记录 5. 更新各字段 6. 保存更新 Args: request: HttpRequest对象，包含文章分类ID和更新信息的JSON数据 Returns: R: 操作结果（成功/失败） """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) # 验证文章分类ID instance_id = data.get('id') if not instance_id or not str(instance_id).isdigit() or int(instance_id) <= 0: return R.failed("文章分类ID无效") instance_id = int(instance_id) # 表单验证 form = forms.CategoryForm(data) if not form.is_valid(): return R.failed(msg=regular.get_err(form)) try: # 查询文章分类记录 instance = Category.objects.filter(id=instance_id, is_delete=False).first() if not instance: return R.failed("文章分类不存在") cleaned_data = form.cleaned_data # 更新各字段 instance.category_name = cleaned_data.get('categoryName') instance.category_code = cleaned_data.get('categoryCode') instance.parent_id = cleaned_data.get('parentId') instance.sort = cleaned_data.get('sort') instance.note = cleaned_data.get('note') instance.update_user = get_username(request) instance.save() return R.ok(msg="更新成功") except Exception as e: logging.error(f"更新文章分类异常，ID: {instance_id}, 错误: {str(e)}") return R.failed(msg="更新失败，请稍后重试") # ============================================================= # 删除服务（支持批量） # ============================================================= def delete_categorys(instance_ids): """ 删除文章分类（批量删除） 采用逻辑删除方式（is_delete=True），而非物理删除。 支持逗号分隔的多个ID。 Args: instance_ids: 文章分类ID字符串，多个ID用逗号分隔 例如: "1" 或 "1,2,3" Returns: R: 操作结果，包含删除数量 """ if not instance_ids: return R.failed("记录ID不存在") try: # 解析ID列表 id_list = [int(id_str.strip()) for id_str in instance_ids.split(',') if id_str.strip()] if not id_list: return R.failed("无效的记录ID") # 批量更新删除标识（逻辑删除） updated_count = Category.objects.filter( id__in=id_list, is_delete=False ).update(is_delete=True) return R.ok(msg=f"本次共删除{updated_count}条数据") except ValueError: return R.failed("记录ID格式错误") except Exception as e: logging.error(f"删除文章分类异常，IDs: {instance_ids}, 错误: {str(e)}") return R.failed(msg="删除失败，请稍后重试") # ============================================================= # 数据下拉列表服务 # ============================================================= def get_category_data(): """ 获取所有启用的文章分类列表 通常用于前端下拉选择框等场景。 只返回 status=1（启用）且未删除的记录。 Returns: list: 文章分类列表，每个元素为文章分类模型字典 查询失败时返回空列表 """ try: # 查询启用的文章分类数据列表 # status=1 表示启用，is_delete=False 表示未删除 queryset = Category.objects.filter( is_delete=False, status=1 ).order_by("sort").values() return list(queryset) except Exception as e: logging.error(f"获取文章分类列表异常: {str(e)}") return []