# +====================================================================== # | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] # +---------------------------------------------------------------------- # | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 # +---------------------------------------------------------------------- # | 开源协议: Apache-2.0 # | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 # +---------------------------------------------------------------------- # | 官方网站: https://www.djangoadmin.cn # +---------------------------------------------------------------------- # | 研发团队: 小蚂蚁云团队 荣誉出品 # +====================================================================== # | 版权和免责声明: # | 一、知识产权声明 # | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 # | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， # | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 # | 二、使用限制与禁止条款 # | 您理解并同意，禁止将本产品用于以下任何行为： # | 1. 任何违反中华人民共和国法律法规的用途； # | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； # | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 # | 三、责任豁免 # | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 # | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 # | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 # | 四、违约与法律责任 # | 如发现任何违反本声明的行为，本团队有权： # | 1. 无条件配合公安机关及其他有权机关的调查取证工作； # | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； # | 3. 保留进一步采取刑事控告等法律措施的权利。 # | 五、许可范围说明 # | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 # | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 # | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 # | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 # | 成的一切经济损失及商誉损害。 # | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 # | 许可协议》附件为准。 # +====================================================================== """ 定时任务日志服务模块 ============================================================= 该模块实现了定时任务日志管理相关的所有业务逻辑。 包括日志的分页查询、详情查询、清理、导出等功能。 定时任务日志用于记录系统中所有定时任务的执行情况，便于监控和排错。 功能模块： 1. 分页查询：支持按任务名称、状态、时间范围等条件筛选 2. 详情查询：根据ID获取单条日志的详细信息 3. 日志清理：支持按时间或条件批量清理历史日志 4. 统计功能：统计任务执行成功率、平均耗时等 5. 数据导出：导出日志数据用于分析 """ import logging from datetime import datetime, timedelta from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage, EmptyPage from application.job_log.models import JobLog from constant.constants import PAGE_SIZE from utils import R from utils.common import format_datetime, parse_request_body # ============================================================= # 分页查询服务 # ============================================================= def get_job_log_page(request): """ 查询定时任务日志分页数据 Args: request: HttpRequest对象，包含查询参数 - pageNo: 当前页码（可选，默认为1） - pageSize: 每页记录数（可选，默认为PAGE_SIZE） - jobName: 任务名称（可选，模糊查询） - jobGroup: 任务分组（可选，精确匹配） - status: 执行状态（可选，0-正常 1-失败） - startTime: 开始时间范围起始（可选） - endTime: 开始时间范围结束（可选） Returns: R: 包含分页数据的JSON响应 """ try: # 获取分页参数 page_no = int(request.GET.get('pageNo', 1)) page_size = int(request.GET.get('pageSize', PAGE_SIZE)) # 构建基础查询（只查询未删除的） query = JobLog.objects.filter(is_delete=False) # 应用筛选条件 query = _apply_filters(query, request) # 排序（默认按创建时间倒序） query = query.order_by("-create_time") # 分页查询 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: 应用筛选后的查询集 """ # 任务名称模糊筛选 job_name = request.GET.get('jobName') if job_name: queryset = queryset.filter(job_name__contains=job_name) # 任务分组精确筛选 job_group = request.GET.get('jobGroup') if job_group: queryset = queryset.filter(job_group=job_group) # 执行状态筛选 status = request.GET.get('status') if status and status.isdigit(): queryset = queryset.filter(status=int(status)) # 任务ID筛选 job_id = request.GET.get('jobId') if job_id and job_id.isdigit(): queryset = queryset.filter(job_id=int(job_id)) # 开始时间范围筛选 start_time = request.GET.get('startTime') if start_time: try: queryset = queryset.filter(start_time__gte=start_time) except Exception: pass end_time = request.GET.get('endTime') if end_time: try: # 将结束日期加一天，包含当天的所有记录 end_date = datetime.strptime(end_time, '%Y-%m-%d') next_day = end_date + timedelta(days=1) queryset = queryset.filter(start_time__lt=next_day) except Exception: pass return queryset def _build_page_data(page_list, paginator, page_no, page_size): """ 构建分页返回数据 Args: page_list: 当前页数据列表 paginator: 分页器对象 page_no: 当前页码 page_size: 每页大小 Returns: dict: 分页数据 """ records = [] for item in page_list: records.append({ 'id': item.id, 'jobId': item.job_id, 'jobName': item.job_name, 'jobGroup': item.job_group, 'jobTrigger': item.job_trigger, 'jobMessage': item.job_message, 'cronExpression': item.cron_expression, 'status': item.status, 'startTime': format_datetime(item.start_time), 'endTime': format_datetime(item.end_time), 'consumeTime': item.consume_time, 'exceptionInfo': item.exception_info, 'createTime': format_datetime(item.create_time), }) return { 'records': records, 'total': paginator.count, 'size': page_size, 'current': page_no, 'pages': paginator.num_pages } # ============================================================= # 详情查询服务 # ============================================================= def get_job_log_detail(log_id): """ 根据ID查询定时任务日志详情 Args: log_id: 日志ID Returns: dict/None: 日志详情或None """ try: log = JobLog.objects.filter(is_delete=False, id=log_id).first() if not log: return None return { 'id': log.id, 'jobId': log.job_id, 'jobName': log.job_name, 'jobGroup': log.job_group, 'jobTrigger': log.job_trigger, 'jobMessage': log.job_message, 'cronExpression': log.cron_expression, 'status': log.status, 'startTime': format_datetime(log.start_time), 'endTime': format_datetime(log.end_time), 'consumeTime': log.consume_time, 'exceptionInfo': log.exception_info, 'createUser': log.create_user, 'createTime': format_datetime(log.create_time), } except Exception as e: logging.error(f"查询日志详情异常，ID: {log_id}, 错误: {str(e)}") return None # ============================================================= # 日志清理服务 # ============================================================= def clean_job_logs(request): """ 清理定时任务日志（支持按条件清理） Args: request: HttpRequest对象，包含清理条件 - cleanType: 清理类型（1-保留最近30天，2-保留最近60天，3-保留最近90天，4-全部清理） - jobId: 指定任务ID（可选） - status: 指定状态（可选） Returns: R: 操作结果 """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) try: clean_type = data.get('cleanType') if not clean_type: return R.failed("请选择清理类型") # 构建查询条件 query = JobLog.objects.filter(is_delete=False) # 按清理类型设置时间条件 if clean_type in [1, 2, 3]: days_map = {1: 30, 2: 60, 3: 90} days = days_map.get(clean_type) cutoff_date = datetime.now() - timedelta(days=days) query = query.filter(create_time__lt=cutoff_date) elif clean_type == 4: # 全部清理，不添加时间条件 pass else: return R.failed("无效的清理类型") # 可选：按任务ID筛选 job_id = data.get('jobId') if job_id and str(job_id).isdigit(): query = query.filter(job_id=int(job_id)) # 可选：按状态筛选 status = data.get('status') if status and str(status).isdigit(): query = query.filter(status=int(status)) # 统计要清理的记录数 clean_count = query.count() if clean_count == 0: return R.ok(msg="没有符合条件的日志需要清理") # 执行清理（逻辑删除） updated_count = query.update(is_delete=True) return R.ok(msg=f"共清理{updated_count}条日志记录") except Exception as e: logging.error(f"清理定时任务日志异常: {str(e)}") return R.failed(msg="清理失败，请稍后重试") # ============================================================= # 删除服务（支持批量） # ============================================================= def delete_job_logs(log_ids): """ 批量删除定时任务日志（逻辑删除） Args: log_ids: 日志ID字符串，多个ID用逗号分隔 Returns: R: 操作结果 """ if not log_ids: return R.failed("记录ID不存在") try: # 解析ID列表 id_list = [int(id_str.strip()) for id_str in log_ids.split(',') if id_str.strip()] if not id_list: return R.failed("无效的记录ID") # 批量逻辑删除 updated_count = JobLog.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: {log_ids}, 错误: {str(e)}") return R.failed(msg="删除失败，请稍后重试") # ============================================================= # 日志导出服务 # ============================================================= def export_job_logs(request): """ 导出定时任务日志数据 Args: request: HttpRequest对象，包含筛选条件 Returns: HttpResponse: CSV文件响应 """ try: # 构建查询 query = JobLog.objects.filter(is_delete=False) # 应用筛选条件 query = _apply_filters(query, request) # 限制导出数量 max_export = 10000 if query.count() > max_export: return R.failed(f"导出数据不能超过{max_export}条，请缩小筛选范围") # 排序 query = query.order_by("-create_time") # 构建导出数据 import csv from django.http import HttpResponse response = HttpResponse(content_type='text/csv') response[ 'Content-Disposition'] = f'attachment; filename="job_logs_{datetime.now().strftime("%Y%m%d%H%M%S")}.csv"' writer = csv.writer(response) writer.writerow(['日志ID', '任务ID', '任务名称', '任务分组', '状态', '开始时间', '结束时间', '耗时(ms)', '日志信息', 'Cron表达式']) for log in query: writer.writerow([ log.id, log.job_id, log.job_name, log.job_group or '', '成功' if log.status == 0 else '失败', format_datetime(log.start_time) or '', format_datetime(log.end_time) or '', log.consume_time, log.job_message or '', log.cron_expression or '' ]) return response except Exception as e: logging.error(f"导出日志数据异常: {str(e)}") return R.failed(msg="导出失败，请稍后重试")