package handler import ( "bytes" "encoding/json" "fmt" "github.com/gin-gonic/gin" "net/http" "os/exec" "strconv" "strings" "time" "videoplayer/dao" "videoplayer/proto" "videoplayer/service" "videoplayer/worker" ) type CIDCreateReq struct { Name string `json:"name" form:"name"` Url string `json:"url" form:"url"` Time int `json:"time" form:"time"` // 定时任务,单位秒,大于0表示定时任务 Script string `json:"script" form:"script"` } type CIDDeleteReq struct { ID int `json:"id" form:"id"` } type CIDRunReq struct { ID int `json:"id" form:"id"` } type CIDLogReq struct { ID int `json:"id" form:"id"` } type CIDUpdateReq struct { ID int `json:"id" form:"id"` Name string `json:"name" form:"name"` Url string `json:"url" form:"url"` Time int `json:"time" form:"time"` // 定时任务,单位秒,大于0表示定时任务 Script string `json:"script" form:"script"` Token string `json:"cidtoken" form:"cidtoken"` } // 全局变量,记录是否进行cron定时任务的刷新 var cron_count int func SetUpCIDGroup(router *gin.Engine) { cidGroup := router.Group("/cid") //持续集成、部署 cidGroup.POST("/create", CreateCID) cidGroup.POST("/delete", DeleteCID) cidGroup.POST("/update", UpdateCID) cidGroup.POST("/list", GetCIDList) cidGroup.POST("/run", RunCID) cidGroup.POST("/log", GetCIDLogList) //获取执行日志 cidGroup.POST("/log/detail", GetCIDLog) //获取执行日志详情 cidGroup.GET("/callback", CIDCallback) cidGroup.POST("/callback", CIDCallback) cidGroup.GET("/running", GetRunningCIDs) } func GetRunningCIDs(c *gin.Context) { id, _ := c.Get("id") user_id := int(id.(float64)) req_type := c.Query("type") //请求方式 0默认自己, 1为所有(管理员可选) req_type_ := 0 if req_type != "" { req_type_, _ = strconv.Atoi(req_type) } resp_data, err := service.GetCIDRunningList(user_id, req_type_) var resp proto.GeneralResp if err != nil { resp.Code = proto.InternalServerError resp.Message = err.Error() } else { resp.Code, resp.Message = 0, "" resp.Data = resp_data } c.JSON(http.StatusOK, resp) } func RunCID(c *gin.Context) { var req CIDRunReq id, _ := c.Get("id") authID := int(id.(float64)) //获取权限 //user := dao.FindUserByUserID(authID) user := service.GetUserByIDFromUserCenter(authID) if user.Run == false { c.JSON(200, gin.H{"error": "no run Permissions", "code": proto.NoRunPermissions, "message": "no run Permissions"}) return } if err := c.ShouldBind(&req); err == nil { // 获取用户ID cid := dao.FindCIDByID(req.ID, authID) if cid.ID == 0 { c.JSON(200, gin.H{"error": "CID not found", "code": proto.OperationFailed, "message": "failed"}) return } else { go RunShellCID(cid.Name, cid.Url, cid.Script, req.ID, authID) c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": "success"}) } } else { c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "failed"}) } } func CreateCID(c *gin.Context) { var req CIDCreateReq if err := c.ShouldBind(&req); err == nil { // 获取用户ID id, _ := c.Get("id") authID := int(id.(float64)) token, _ := generateRandomHexString(32) res := dao.CreateCID(req.Name, req.Url, req.Script, token, req.Time, authID) if res != 0 { if req.Time > 0 { updateCronRedisTime(int(res), req.Time) } c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res}) } else { c.JSON(200, gin.H{"error": "CreateCID failed", "code": proto.OperationFailed, "message": "failed"}) } } else { c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "failed"}) } } func DeleteCID(c *gin.Context) { var req CIDDeleteReq if err := c.ShouldBind(&req); err == nil { // 获取用户ID id, _ := c.Get("id") authID := int(id.(float64)) cid := dao.DeleteCIDByID(req.ID, authID) if cid == false { c.JSON(200, gin.H{"error": "CID not found", "code": proto.OperationFailed, "message": "failed"}) return } else { c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": "success"}) } } else { c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "failed"}) } } func UpdateCID(c *gin.Context) { var req CIDUpdateReq if err := c.ShouldBind(&req); err == nil { // 获取用户ID id, _ := c.Get("id") authID := int(id.(float64)) cid := dao.UpdateCIDByID(req.ID, authID, req.Time, req.Name, req.Url, req.Script, req.Token) if cid == false { c.JSON(200, gin.H{"error": "CID not found", "code": proto.OperationFailed, "message": "failed"}) return } else { if req.Time > 0 { updateCronRedisTime(req.ID, req.Time) } c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": "success"}) } } else { c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "failed"}) } } func GetCIDList(c *gin.Context) { // 获取用户ID id, _ := c.Get("id") authID := int(id.(float64)) cids := dao.FindCIDByAuthID(authID) c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": cids}) } func GetCIDLog(c *gin.Context) { var req CIDRunReq if err := c.ShouldBind(&req); err == nil { // 获取用户ID id, _ := c.Get("id") authID := int(id.(float64)) cidLogs := dao.FindRunLogByCIDLogID(req.ID, authID) c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": cidLogs}) } else { c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "failed"}) } } func GetCIDLogList(c *gin.Context) { // 获取用户ID id, _ := c.Get("id") authID := int(id.(float64)) var req CIDLogReq if err := c.ShouldBind(&req); err == nil { cidLogs := dao.FindRunLogByID(authID, req.ID) c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": cidLogs}) } else { c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "failed"}) } } func CIDCallback(c *gin.Context) { // 获取用户ID token := c.Query("token") cid_id := c.Query("id") //将cid转换为int cid, _ := strconv.Atoi(cid_id) var req proto.CIDCallBackReq var resp proto.GeneralResp if err := c.ShouldBindQuery(&req); err != nil { resp.Code, resp.Message = proto.ParameterError, err.Error() c.JSON(http.StatusOK, resp) return } if req.ID == 0 || req.Token == "" { resp.Code, resp.Message = proto.ParameterError, "token or id is empty" c.JSON(http.StatusOK, resp) return } res := dao.FindCIDByIDAndToken(cid, token) if res.ID == 0 { resp.Code, resp.Message = proto.ParameterError, "CID not found by id and token:"+req.Token+", id:"+strconv.Itoa(int(res.ID)) c.JSON(http.StatusOK, resp) return } //user := dao.FindUserByUserID(res.Auth_id) user := service.GetUserByIDFromUserCenter(res.Auth_id) if user.Run == false { resp.Code, resp.Message = proto.NoRunPermissions, "the user has no run Permissions" c.JSON(http.StatusOK, resp) return } if res.ID != 0 { go RunShellCID(res.Name, res.Url, res.Script, int(res.ID), res.Auth_id) resp.Code, resp.Message, resp.Data = proto.SuccessCode, "success", res.Name c.JSON(http.StatusOK, resp) return } else { resp.Code, resp.Message = proto.OperationFailed, "CID not found by id and token" c.JSON(http.StatusOK, resp) return } } func RunShell(username, url, script string, id, authID int) { strs := strings.Split(url, "/") name := strs[len(strs)-1] names := strings.Split(name, ".") name = names[0] //脚本内容,不同用户的持续集成、部署目录不同 scriptContent := ` echo "start" ` + script + ` echo "end"` start := time.Now() //执行脚本 cmd := exec.Command("/bin/bash", "-c", scriptContent) // 使用bytes.Buffer捕获输出 var out bytes.Buffer cmd.Stdout = &out err3 := cmd.Run() err3_info := "" if err3 != nil { err3_info = err3.Error() } elapsed := time.Since(start) //fmt.Println("bash content:", scriptContent) dao.CreateRunLog(id, authID, scriptContent, out.String(), err3_info, elapsed.Seconds()) //添加执行日志 } func RunShellCID(cid_name, url, script string, id, authID int) { strs := strings.Split(url, "/") name := strs[len(strs)-1] names := strings.Split(name, ".") name = names[0] now := time.Now() var cid_running proto.CIDRunning cid_running.ID = id cid_running.AuthID = authID cid_running.StartTime = now cid_running.CID = cid_name //加入正在运行 proto.CID_RunningMutex.Lock() user_running_list := proto.CID_Running_Map[authID] user_running_list = append(user_running_list, cid_running) proto.CID_Running_Map[authID] = user_running_list proto.CID_RunningMutex.Unlock() //脚本内容,不同用户的持续集成、部署目录不同 scriptContent := ` echo "start" ` + script + ` echo "end"` start := time.Now() //执行脚本 cmd := exec.Command("/bin/bash", "-c", scriptContent) // 使用bytes.Buffer捕获输出 var out bytes.Buffer cmd.Stdout = &out err3 := cmd.Run() err3_info := "" if err3 != nil { err3_info = err3.Error() } elapsed := time.Since(start) //fmt.Println("bash content:", scriptContent) dao.CreateRunLog(id, authID, scriptContent, out.String(), err3_info, elapsed.Seconds()) //添加执行日志 //移除正在运行 proto.CID_RunningMutex.Lock() user_running_list = proto.CID_Running_Map[authID] for i, v := range user_running_list { if v.StartTime.Equal(now) == true { //删除 user_running_list = append(user_running_list[:i], user_running_list[i+1:]...) break } } proto.CID_Running_Map[authID] = user_running_list proto.CID_RunningMutex.Unlock() } // 定时任务处理逻辑 func RunCron() { cron_count++ if cron_count > 6 { updateCronFromDBToRedis() cron_count = 0 } //从redis查看是否有定时任务 //如果有定时任务,执行定时任务 //如果没有定时任务,查找数据库是否有定时任务,如果有,则加入redis,如果没有,则不做任何操作 key := "cron_cid_runs" res := worker.GetRedis(key) //fmt.Println("cid run cron res:", res) if res == "" { readCronFromDBToRedis(key) } else { var cid_workers []proto.CIDRUN err := json.Unmarshal([]byte(res), &cid_workers) if err != nil { fmt.Println("json unmarshal failed") } //fmt.Println("cid_workers:", cid_workers) for i, v := range cid_workers { //查找定时任务 if v.Curr-10 <= 0 { cid := dao.FindCIDByCID(v.CID) if cid.ID != 0 { go RunShell("cron", cid.Url, cid.Script, int(cid.ID), cid.Auth_id) } cid_workers[i].Curr = v.Every } else { cid_workers[i].Curr = v.Curr - 10 } } //将定时任务加入redis json_data, err := json.Marshal(cid_workers) if err != nil { fmt.Println("json marshal failed") } worker.SetRedis(key, string(json_data)) } } // 将数据库中的定时任务加入redis func readCronFromDBToRedis(key string) { cids := dao.FindCIDByTime() cid_workers := make([]proto.CIDRUN, 0) for _, v := range cids { cid_worker := proto.CIDRUN{CID: v.ID, Curr: v.Time, Every: v.Time} cid_workers = append(cid_workers, cid_worker) } if len(cid_workers) > 0 { //将定时任务加入redis json_data, err := json.Marshal(cid_workers) if err != nil { fmt.Println("json marshal failed") } worker.SetRedis(key, string(json_data)) } } // 从数据库更新定时任务到redis func updateCronFromDBToRedis() { key := "cron_cid_runs" cids := dao.FindCIDByTime() cid_maps := make(map[uint]int) //将数据库中的定时任务加入map,key为cid,value为时间,便于后续查找 for _, v := range cids { cid_maps[v.ID] = v.Time } res := worker.GetRedis(key) if res == "" { readCronFromDBToRedis(key) return } var cid_workers_redis []proto.CIDRUN err := json.Unmarshal([]byte(res), &cid_workers_redis) if err != nil { fmt.Println("json unmarshal failed") return } for i, v := range cid_workers_redis { if time, ok := cid_maps[v.CID]; ok { if v.Every != time { cid_workers_redis[i].Every = time cid_workers_redis[i].Curr = time } } } //将定时任务加入redis json_data, err := json.Marshal(cid_workers_redis) if err != nil { fmt.Println("json marshal failed") return } worker.SetRedis(key, string(json_data)) } // 查看指定定时任务是否存在,如果存在则更新时间,如果不存在则加入 func updateCronRedisTime(id int, time int) { key := "cron_cid_runs" res := worker.GetRedis(key) if res == "" { readCronFromDBToRedis(key) return } else { var cid_workers []proto.CIDRUN err := json.Unmarshal([]byte(res), &cid_workers) if err != nil { fmt.Println("json unmarshal failed") } isContain := false for i, v := range cid_workers { if v.CID == uint(id) { //更新时间,不会继续原来的时间 cid_workers[i].Curr = time cid_workers[i].Every = time isContain = true } } if isContain == false { cid_worker := proto.CIDRUN{CID: uint(id), Curr: time, Every: time} cid_workers = append(cid_workers, cid_worker) } //将定时任务加入redis json_data, err := json.Marshal(cid_workers) if err != nil { fmt.Println("json marshal failed") } worker.SetRedis(key, string(json_data)) } }