package controller import ( "context" "fmt" "ppgo_job/consts" "ppgo_job/dao" "ppgo_job/model/entity" "ppgo_job/service/scheduler" "strings" "time" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/ghttp" "github.com/gogf/gf/v2/os/gcron" "github.com/gogf/gf/v2/util/gconv" ) // ===== 任务管理 ===== func TaskTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() groupId := r.Get("group_id", -1).Int() serverName := strings.TrimSpace(r.Get("serverName", "").String()) status := r.Get("status", -1).Int() taskName := strings.TrimSpace(r.Get("task_name", "").String()) var filters []interface{} if groupId >= 0 { filters = append(filters, "group_id", groupId) } if serverName != "" { filters = append(filters, "server_name like", "%"+serverName+"%") } if status >= 0 { filters = append(filters, "status", status) } else { filters = append(filters, "status >", -1) } if taskName != "" { filters = append(filters, "task_name like", "%"+taskName+"%") } if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.Task.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } // 计算各任务的下次执行时间(基于 cron 表达式) nextTimeMap := make(map[int]string) for _, t := range list { if t.Status == 1 && t.CronSpec != "" { if nt := scheduler.GetTaskNextTime(t.CronSpec); nt != "" { nextTimeMap[t.Id] = nt } } } rows := make([]g.Map, 0) for _, t := range list { st := "运行中" if t.Status == -1 { st = "已删除" } else if t.Status == 0 { st = "已暂停" } else if t.Status == 2 { st = "待审核" } else if t.Status == 3 { st = "审核失败" } nt := nextTimeMap[t.Id] pt := "" if t.PrevTime > 0 { pt = time.Unix(t.PrevTime, 0).Format("2006-01-02 15:04:05") } ct := "" ut := "" if t.CreateTime > 0 { ct = time.Unix(t.CreateTime, 0).Format("2006-01-02 15:04:05") } if t.UpdateTime > 0 { ut = time.Unix(t.UpdateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": t.Id, "task_name": t.TaskName, "group_id": t.GroupId, "cron_spec": t.CronSpec, "task_type": t.TaskType, "command": t.Command, "url": t.Url, "method": t.Method, "headers": t.Headers, "body": t.Body, "status": t.Status, "status_text": st, "prev_time": t.PrevTime, "prev_time_str": pt, "next_time": nt, "execute_times": t.ExecuteTimes, "server_ids": t.ServerIds, "create_time_str": ct, "update_time_str": ut}) } ajaxList(r, "成功", 0, total, rows) } func TaskAjaxSave(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() taskName := r.Get("task_name").String() cronSpec := r.Get("cron_spec").String() taskType := r.Get("task_type", "shell").String() command := r.Get("command").String() url := r.Get("url").String() method := r.Get("method", "GET").String() headers := r.Get("headers").String() body := r.Get("body").String() groupId := r.Get("group_id", 0).Int() serverIds := r.Get("server_ids").String() serverType := r.Get("server_type", 0).Int() concurrent := r.Get("concurrent", 0).Int() timeout := r.Get("timeout", 0).Int() isNotify := r.Get("is_notify", 0).Int() notifyType := r.Get("notify_type", 0).Int() notifyTplId := r.Get("notify_tpl_id", 0).Int() notifyUserIds := r.Get("notify_user_ids").String() description := r.Get("description").String() if taskName == "" || cronSpec == "" { ajaxMsg(r, "任务名称、Cron表达式不能为空", consts.MsgErr) return } if taskType == "http" { if url == "" { ajaxMsg(r, "URL不能为空", consts.MsgErr) return } } else { if command == "" { ajaxMsg(r, "命令不能为空", consts.MsgErr) return } if err := scheduler.Scheduler.CheckCommand(ctx, command); err != nil { ajaxMsg(r, err.Error(), consts.MsgErr) return } } // 验证 Cron 表达式(临时注册再移除,仅用于校验格式) if _, err := gcron.AddSingleton(ctx, cronSpec, func(ctx context.Context) {}, "cron_validator"); err != nil { ajaxMsg(r, "Cron表达式格式错误: "+err.Error(), consts.MsgErr) return } gcron.Remove("cron_validator") now := time.Now().Unix() userId := r.GetCtxVar("userId", 0).Int() if id > 0 { task, _ := dao.Task.GetById(ctx, id) if task == nil { ajaxMsg(r, "任务不存在", consts.MsgErr) return } task.TaskName = taskName task.Description = description task.GroupId = groupId task.ServerIds = serverIds task.ServerType = serverType task.CronSpec = cronSpec task.Concurrent = concurrent task.TaskType = taskType task.Command = command task.Url = url task.Method = method task.Headers = headers task.Body = body task.Timeout = timeout task.IsNotify = isNotify task.NotifyType = notifyType task.NotifyTplId = notifyTplId task.NotifyUserIds = notifyUserIds task.UpdateId = userId task.UpdateTime = now if err := dao.Task.Update(ctx, task); err != nil { ajaxMsg(r, "更新失败: "+err.Error(), consts.MsgErr) return } if task.Status == 1 { scheduler.Scheduler.UpdateTask(ctx, task) } ajaxMsg(r, "更新成功", consts.MsgOK) } else { status := 2 if userId == 1 { status = 0 } t := &entity.Task{ TaskName: taskName, Description: description, GroupId: groupId, ServerIds: serverIds, ServerType: serverType, CronSpec: cronSpec, Concurrent: concurrent, TaskType: taskType, Command: command, Url: url, Method: method, Headers: headers, Body: body, Timeout: timeout, Status: status, IsNotify: isNotify, NotifyType: notifyType, NotifyTplId: notifyTplId, NotifyUserIds: notifyUserIds, CreateId: userId, UpdateId: userId, CreateTime: now, UpdateTime: now, } id, err := dao.Task.Insert(ctx, t) if err != nil { ajaxMsg(r, "添加失败: "+err.Error(), consts.MsgErr) return } ajaxMsg(r, id, consts.MsgOK) } } func TaskAjaxDel(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } task, _ := dao.Task.GetById(ctx, id) if task == nil { ajaxMsg(r, "任务不存在", consts.MsgErr) return } _ = dao.Task.Delete(ctx, id) if task.Status == 1 { scheduler.Scheduler.RemoveTask(ctx, id) } ajaxMsg(r, "删除成功", consts.MsgOK) } func TaskAjaxStart(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 1}) task, _ := dao.Task.GetById(ctx, id) if task != nil { scheduler.Scheduler.AddTask(ctx, task) } ajaxMsg(r, "启动成功", consts.MsgOK) } func TaskAjaxPause(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 0}) scheduler.Scheduler.RemoveTask(ctx, id) ajaxMsg(r, "暂停成功", consts.MsgOK) } func TaskAjaxRun(r *ghttp.Request) { id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = scheduler.Scheduler.RunTaskNow(r.GetCtx(), id) ajaxMsg(r, "任务已触发执行", consts.MsgOK) } func TaskAjaxAudit(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 1}) task, _ := dao.Task.GetById(ctx, id) if task != nil { scheduler.Scheduler.AddTask(ctx, task) } ajaxMsg(r, "审核通过,任务已启动", consts.MsgOK) } func TaskAjaxNopass(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 3}) ajaxMsg(r, "已设置为审核失败", consts.MsgOK) } func TaskAjaxBatchStart(r *ghttp.Request) { ctx := r.GetCtx() ids := r.Get("ids").String() for _, idStr := range strings.Split(ids, ",") { if id := gconv.Int(strings.TrimSpace(idStr)); id > 0 { _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 1}) if t, _ := dao.Task.GetById(ctx, id); t != nil { scheduler.Scheduler.AddTask(ctx, t) } } } ajaxMsg(r, "批量启动成功", consts.MsgOK) } func TaskAjaxBatchPause(r *ghttp.Request) { ctx := r.GetCtx() ids := r.Get("ids").String() for _, idStr := range strings.Split(ids, ",") { if id := gconv.Int(strings.TrimSpace(idStr)); id > 0 { _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 0}) scheduler.Scheduler.RemoveTask(ctx, id) } } ajaxMsg(r, "批量暂停成功", consts.MsgOK) } func TaskAjaxBatchDel(r *ghttp.Request) { ctx := r.GetCtx() ids := r.Get("ids").String() for _, idStr := range strings.Split(ids, ",") { if id := gconv.Int(strings.TrimSpace(idStr)); id > 0 { _ = dao.Task.Delete(ctx, id) scheduler.Scheduler.RemoveTask(ctx, id) } } ajaxMsg(r, "批量删除成功", consts.MsgOK) } func TaskAjaxBatchAudit(r *ghttp.Request) { ctx := r.GetCtx() ids := r.Get("ids").String() for _, idStr := range strings.Split(ids, ",") { if id := gconv.Int(strings.TrimSpace(idStr)); id > 0 { _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 1}) if t, _ := dao.Task.GetById(ctx, id); t != nil { scheduler.Scheduler.AddTask(ctx, t) } } } ajaxMsg(r, "批量审核通过", consts.MsgOK) } func TaskAjaxBatchNoPass(r *ghttp.Request) { ctx := r.GetCtx() ids := r.Get("ids").String() for _, idStr := range strings.Split(ids, ",") { if id := gconv.Int(strings.TrimSpace(idStr)); id > 0 { _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 3}) } } ajaxMsg(r, "批量审核不通过", consts.MsgOK) } func TaskApiTask(r *ghttp.Request) { ajaxMsg(r, "API接口待完善", consts.MsgOK) } func TaskApiStart(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 1}) if t, _ := dao.Task.GetById(ctx, id); t != nil { scheduler.Scheduler.AddTask(ctx, t) } ajaxMsg(r, id, consts.MsgOK) } func TaskApiPause(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.Task.UpdateFields(ctx, id, g.Map{"status": 0}) scheduler.Scheduler.RemoveTask(ctx, id) ajaxMsg(r, id, consts.MsgOK) } func TaskLogAjaxDel(r *ghttp.Request) { ctx := r.GetCtx() id := r.Get("id", 0).Int() if id <= 0 { ajaxMsg(r, "参数错误", consts.MsgErr) return } _ = dao.TaskLog.DeleteById(ctx, id) ajaxMsg(r, "删除成功", consts.MsgOK) } // ===== 服务器管理 ===== func ServerTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() groupId := r.Get("group_id", -1).Int() serverName := strings.TrimSpace(r.Get("serverName", "").String()) var filters []interface{} if groupId >= 0 { filters = append(filters, "group_id", groupId) } if serverName != "" { filters = append(filters, "server_name like", "%"+serverName+"%") } if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.TaskServer.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } // 获取服务器分组名称 groups, _, _ := dao.ServerGroup.GetList(ctx, 1, 1000) sgMap := make(map[int]string) for _, g := range groups { sgMap[g.Id] = g.GroupName } rows := make([]g.Map, 0) for _, s := range list { statusText := "正常" if s.Status == 0 { statusText = "禁用" } connType := "本地" if s.ConnectionType == 1 { connType = "Telnet" } else if s.ConnectionType == 2 { connType = "Agent" } sgName := sgMap[s.GroupId] ct := "" ut := "" if s.CreateTime > 0 { ct = time.Unix(s.CreateTime, 0).Format("2006-01-02 15:04:05") } if s.UpdateTime > 0 { ut = time.Unix(s.UpdateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": s.Id, "server_name": s.ServerName, "server_ip": s.ServerIp, "port": s.Port, "status": s.Status, "statusText": statusText, "connTypeText": connType, "connection_type": s.ConnectionType, "group_name": sgName, "detail": s.Detail, "create_time_str": ct, "update_time_str": ut}) } ajaxList(r, "成功", 0, total, rows) } // ===== 任务分组 ===== func GroupTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() var filters []interface{} if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.Group.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } rows := make([]g.Map, 0) for _, v := range list { ct := "" ut := "" if v.CreateTime > 0 { ct = time.Unix(v.CreateTime, 0).Format("2006-01-02 15:04:05") } if v.UpdateTime > 0 { ut = time.Unix(v.UpdateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": v.Id, "group_name": v.GroupName, "description": v.Description, "create_time": ct, "update_time": ut, "status": v.Status}) } ajaxList(r, "成功", 0, total, rows) } func ServerGroupTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() var filters []interface{} if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.ServerGroup.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } rows := make([]g.Map, 0) for _, v := range list { ct := "" ut := "" if v.CreateTime > 0 { ct = time.Unix(v.CreateTime, 0).Format("2006-01-02 15:04:05") } if v.UpdateTime > 0 { ut = time.Unix(v.UpdateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": v.Id, "group_name": v.GroupName, "description": v.Description, "create_time": ct, "update_time": ut, "status": v.Status}) } ajaxList(r, "成功", 0, total, rows) } // ===== 执行日志 ===== func TaskLogTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() taskId := r.Get("task_id", 0).Int() status := r.Get("status", 9).Int() var filters []interface{} if taskId > 0 { filters = append(filters, "task_id", taskId) } if status != 9 { filters = append(filters, "status", status) } if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.TaskLog.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } // 构建任务ID→名称映射 taskNameMap := make(map[int]string) for _, l := range list { if _, ok := taskNameMap[l.TaskId]; !ok { if t, _ := dao.Task.GetById(ctx, l.TaskId); t != nil { taskNameMap[l.TaskId] = t.TaskName } } } rows := make([]g.Map, 0) for _, l := range list { st := "成功" if l.Status == -1 { st = "失败" } else if l.Status == -2 { st = "超时" } startTime := "" if l.CreateTime > 0 { startTime = time.Unix(l.CreateTime, 0).Format("2006-01-02 15:04:05") } outputSize := len(l.Output) sizeText := fmt.Sprintf("%d B", outputSize) if outputSize >= 1024 { sizeText = fmt.Sprintf("%.1f KB", float64(outputSize)/1024) } rows = append(rows, g.Map{ "id": l.Id, "task_id": l.TaskId, "task_name": taskNameMap[l.TaskId], "server_name": l.ServerName, "status": l.Status, "statusText": st, "process_time": l.ProcessTime, "start_time": startTime, "output_size": sizeText, "output": l.Output, "error": l.Error, }) } ajaxList(r, "成功", 0, total, rows) } func BanTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() var filters []interface{} if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.Ban.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } rows := make([]g.Map, 0) for _, v := range list { ct := "" if v.CreateTime > 0 { ct = time.Unix(v.CreateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": v.Id, "code": v.Code, "create_time": ct, "status": v.Status}) } ajaxList(r, "成功", 0, total, rows) } // ===== 权限管理 ===== func RoleTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() var filters []interface{} if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.Role.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } rows := make([]g.Map, 0) for _, v := range list { ct := "" ut := "" if v.CreateTime > 0 { ct = time.Unix(v.CreateTime, 0).Format("2006-01-02 15:04:05") } if v.UpdateTime > 0 { ut = time.Unix(v.UpdateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": v.Id, "role_name": v.RoleName, "detail": v.Detail, "create_time": ct, "update_time": ut, "status": v.Status}) } ajaxList(r, "成功", 0, total, rows) } func AdminTable(r *ghttp.Request) { ctx := r.GetCtx() page := r.Get("page", 1).Int() limit := r.Get("limit", 20).Int() realName := strings.TrimSpace(r.Get("realName", "").String()) var filters []interface{} if realName != "" { filters = append(filters, "real_name like", "%"+realName+"%") } if sortStr := parseSort(r); sortStr != "" { filters = append(filters, "@order", sortStr) } list, total, err := dao.Admin.GetList(ctx, page, limit, filters...) if err != nil { ajaxList(r, "查询失败", consts.MsgErr, 0, nil) return } rows := make([]g.Map, 0) for _, a := range list { statusText := "正常" if a.Status == 0 { statusText = "禁用" } ct := "" ut := "" if a.CreateTime > 0 { ct = time.Unix(a.CreateTime, 0).Format("2006-01-02 15:04:05") } if a.UpdateTime > 0 { ut = time.Unix(a.UpdateTime, 0).Format("2006-01-02 15:04:05") } rows = append(rows, g.Map{"id": a.Id, "login_name": a.LoginName, "real_name": a.RealName, "phone": a.Phone, "email": a.Email, "status": a.Status, "statusText": statusText, "create_time_str": ct, "update_time_str": ut}) } ajaxList(r, "成功", 0, total, rows) }