videoplayer/handler/cid.go

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 > 0 {
		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 > 0 {
		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()) //添加执行日志
	if err3 != nil {
		dao.UpdateLastFailByID(id, time.Now())
	} else {
		dao.UpdateLastSuccessByID(id, time.Now())
	}
	//移除正在运行
	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))
	}
}