1、增加报表功能；

2、增加服务器端sse；
3、增加任务运行日志；

config/config.yaml

@@ -42,4 +42,7 @@ node:
 functions:
   - name: task-request
     dataTag : 12
-    mqType: 1
+    mqType: 1
+  - name: task-log
+    dataTag: 28
+    mqType: 2

go.mod

@@ -5,7 +5,9 @@ go 1.18
 require (
 	git.hpds.cc/Component/gin_valid v0.0.0-20230104142509-f956bce255b6
 	git.hpds.cc/Component/logging v0.0.0-20230106105738-e378e873921b
-	git.hpds.cc/pavement/hpds_node v0.0.0-20230402152619-41414aafa930
+	git.hpds.cc/Component/network v0.0.0-20230405135741-a4ea724bab76
+	git.hpds.cc/pavement/hpds_node v0.0.0-20230405153516-9403c4d01e12
+	github.com/360EntSecGroup-Skylar/excelize v1.4.1
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/gin-contrib/zap v0.1.0
 	github.com/gin-gonic/gin v1.8.2
@@ -17,6 +19,7 @@ require (
 	github.com/spf13/cobra v0.0.3
 	github.com/spf13/viper v1.14.0
 	go.uber.org/zap v1.24.0
+	golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8
 	gopkg.in/yaml.v3 v3.0.1
 	xorm.io/xorm v1.3.2
 )
@@ -27,7 +30,6 @@ require (
 	cloud.google.com/go/compute/metadata v0.2.1 // indirect
 	cloud.google.com/go/firestore v1.8.0 // indirect
 	git.hpds.cc/Component/mq_coder v0.0.0-20221010064749-174ae7ae3340 // indirect
-	git.hpds.cc/Component/network v0.0.0-20230326151855-3c157f531d86 // indirect
 	github.com/armon/go-metrics v0.4.0 // indirect
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
@@ -45,6 +47,7 @@ require (
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
+	github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.6.0 // indirect
@@ -60,10 +63,7 @@ require (
 	github.com/klauspost/compress v1.15.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.1.0 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
-	github.com/lucas-clemente/quic-go v0.29.1 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
-	github.com/marten-seemann/qtls-go1-18 v0.1.2 // indirect
-	github.com/marten-seemann/qtls-go1-19 v0.1.0 // indirect
 	github.com/matoous/go-nanoid/v2 v2.0.0 // indirect
 	github.com/mattn/go-colorable v0.1.12 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
@@ -73,11 +73,15 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/nxadm/tail v1.4.8 // indirect
+	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect
 	github.com/onsi/ginkgo v1.16.4 // indirect
+	github.com/onsi/ginkgo/v2 v2.2.0 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/quic-go/qtls-go1-19 v0.2.1 // indirect
+	github.com/quic-go/qtls-go1-20 v0.1.1 // indirect
+	github.com/quic-go/quic-go v0.33.0 // indirect
 	github.com/rs/xid v1.4.0 // indirect
 	github.com/sagikazarmark/crypt v0.8.0 // indirect
 	github.com/sirupsen/logrus v1.9.0 // indirect
@@ -97,16 +101,16 @@ require (
 	go.opentelemetry.io/otel/trace v1.10.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
-	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa // indirect
+	golang.org/x/crypto v0.4.0 // indirect
-	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
+	golang.org/x/exp v0.0.0-20221205204356-47842c84f3db // indirect
-	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
+	golang.org/x/mod v0.6.0 // indirect
 	golang.org/x/net v0.4.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783 // indirect
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.3.0 // indirect
 	golang.org/x/text v0.5.0 // indirect
 	golang.org/x/time v0.0.0-20220609170525-579cf78fd858 // indirect
-	golang.org/x/tools v0.1.12 // indirect
+	golang.org/x/tools v0.2.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	google.golang.org/api v0.102.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
@@ -115,7 +119,6 @@ require (
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
-	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	xorm.io/builder v0.3.11-0.20220531020008-1bd24a7dc978 // indirect
 )

internal/handler/dataset.go

@@ -63,3 +63,18 @@ func (s HandlerService) ImportDataset(c *gin.Context) (data interface{}, err err
 	go s.SaveLog("导入数据集", "Dataset", "", "", ToString(data), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 	return
 }
+
+func (s HandlerService) DatasetInfo(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewDatasetService(s.AppConfig, s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.DatasetItemRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("DatasetInfo", "Dataset", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	data, err = repo.DatasetInfo(c, req)
+	go s.SaveLog("获取数据集详情", "Dataset", "", "", ToString(data), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}

internal/handler/file.go

@@ -37,3 +37,27 @@ func (s HandlerService) UploadFile(c *gin.Context) (data interface{}, err error)
 	go s.SaveLog("文件上传", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 	return
 }
+
+func (s HandlerService) FileList(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewFileService(s.AppConfig, s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.DatasetItemRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("FileList", "FileManage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	if req.Size < 1 {
+		req.Size = 20
+	}
+	if req.Size > 1000 {
+		req.Size = 1000
+	}
+	if req.Page < 1 {
+		req.Page = 1
+	}
+	data, err = repo.FileList(c, req)
+	go s.SaveLog("获取数据集详情", "FileManage", "", "", ToString(data), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}

internal/handler/manage.go

@@ -122,6 +122,20 @@ func (s HandlerService) ProjectList(c *gin.Context) (data interface{}, err error
 	return
 }
 
+func (s HandlerService) ProjectInfo(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewManageService(s.AppConfig, s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.ProjectItemRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("ProjectInfo", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	data, err = repo.ProjectInfo(c, req)
+	go s.SaveLog("获取项目信息", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}
 func (s HandlerService) AddProject(c *gin.Context) (data interface{}, err error) {
 	repo := service.NewManageService(s.AppConfig, s.Engine, s.Logger)
 	us, _ := c.Get("operatorUser")

internal/handler/report.go

@@ -0,0 +1,47 @@
+package handler
+
+import (
+	"fmt"
+	"github.com/gin-gonic/gin"
+	"hpds-iot-web/internal/proto"
+	"hpds-iot-web/internal/service"
+	"hpds-iot-web/model"
+	e "hpds-iot-web/pkg/err"
+)
+
+func (s HandlerService) ReportList(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewReportService(s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.ReportRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("ReportList", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	data, err = repo.ReportList(c, req)
+	go s.SaveLog("报表列表", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}
+
+func (s HandlerService) GenerateReport(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewReportService(s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.ReportRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("GenerateReport", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	xlsx, err := repo.GenerateReport(c, req)
+	if err != nil {
+
+	}
+	c.Header("Content-Type", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
+	c.Header("Content-Disposition", "attachment; filename=模型报表.xlsx")
+	//c.Header("")
+	c.Header("Content-Transfer-Encoding", "binary")
+	err = xlsx.Write(c.Writer)
+	return nil, err
+}

internal/handler/task.go

@@ -19,6 +19,15 @@ func (s HandlerService) TaskList(c *gin.Context) (data interface{}, err error) {
 		go s.SaveLog("TaskList", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 		return nil, e.NewValidErr(err)
 	}
+	if req.Size < 1 {
+		req.Size = 20
+	}
+	if req.Size > 1000 {
+		req.Size = 1000
+	}
+	if req.Page < 1 {
+		req.Page = 1
+	}
 	data, err = repo.TaskList(c, req)
 	go s.SaveLog("任务列表", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 	return
@@ -38,3 +47,71 @@ func (s HandlerService) AddTask(c *gin.Context) (data interface{}, err error) {
 	go s.SaveLog("新增任务", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 	return
 }
+func (s HandlerService) ReRunTask(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewTaskService(s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.TaskItemRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("ReRunTask", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	data, err = repo.ReRunTask(c, req)
+	go s.SaveLog("重新执行任务", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}
+
+func (s HandlerService) TaskInfo(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewTaskService(s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.TaskItemRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("TaskInfo", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	data, err = repo.TaskInfo(c, req)
+	go s.SaveLog("获取任务信息", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}
+
+func (s HandlerService) TaskResult(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewTaskService(s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.ReportRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("TaskResult", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	if req.Size < 1 {
+		req.Size = 20
+	}
+	if req.Size > 1000 {
+		req.Size = 1000
+	}
+	if req.Page < 1 {
+		req.Page = 1
+	}
+	data, err = repo.TaskResult(c, req)
+	go s.SaveLog("获取任务结果信息", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}
+
+func (s HandlerService) TaskLog(c *gin.Context) (data interface{}, err error) {
+	repo := service.NewTaskService(s.Engine, s.Logger)
+	us, _ := c.Get("operatorUser")
+	userInfo := us.(*model.SystemUser)
+	var req proto.TaskItemRequest
+	err = c.ShouldBindJSON(&req)
+	if err != nil {
+		go s.SaveLog("TaskLog", "Manage", "", "", req.ToString(), fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	data, err = repo.TaskLog(c, req)
+	go s.SaveLog("获取任务日志信息", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
+	return
+}

internal/handler/user.go

@@ -33,7 +33,6 @@ func (s HandlerService) AddUser(c *gin.Context) (data interface{}, err error) {
 	data, err = repo.AddUser(c, req)
 	go s.SaveLog("新增用户", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 	return
-	return
 }
 
 func (s HandlerService) EditUser(c *gin.Context) (data interface{}, err error) {
@@ -49,7 +48,6 @@ func (s HandlerService) EditUser(c *gin.Context) (data interface{}, err error) {
 	data, err = repo.EditUser(c, req)
 	go s.SaveLog("修改用户信息", "Manage", "", "", "", fmt.Sprintf("%d", userInfo.UserId), c.Request.RemoteAddr, "")
 	return
-	return
 }
 
 func (s HandlerService) GetUserInfo(c *gin.Context) (data interface{}, err error) {
@@ -80,5 +78,10 @@ func (s HandlerService) MenuList(c *gin.Context) (data interface{}, err error) {
 	us, _ := c.Get("operatorUser")
 	userinfo := us.(*model.SystemUser)
 	data, err = repo.MenuList(c, userinfo.UserId)
+	if err != nil {
+		go s.SaveLog("GetUserList", "Manage", "", "", userinfo.ToString(), fmt.Sprintf("%d", userinfo.UserId), c.Request.RemoteAddr, "")
+		return nil, e.NewValidErr(err)
+	}
+	go s.SaveLog("获取用户菜单列表", "Manage", "", "", "", fmt.Sprintf("%d", userinfo.UserId), c.Request.RemoteAddr, "")
 	return
 }

internal/middleware/jwt.go

@@ -70,7 +70,7 @@ func JwtAuthMiddleware(logger *zap.Logger) gin.HandlerFunc {
 		)
 		token := c.GetHeader("Authorization")
 		// 这里可以过滤不需要进行验证的接口
-		if path == "/api/user/login" || path == "/api/health" {
+		if path == "/api/user/login" || path == "/api/health" || path == "/api/task/event" {
 			goto Return
 		}
 		if len(token) == 0 {

internal/proto/request.go

@@ -428,6 +428,11 @@ type UploadFileRequest struct {
 	Scene     string                  `json:"scene"`
 	Files     []*multipart.FileHeader `json:"files"`
 	DatasetId int64                   `json:"datasetId"`
+	DataType  int                     `json:"dataType"`
+}
+
+type FileList struct {
+	DatasetId int64
 }
 
 type BrandRequest struct {
@@ -513,11 +518,11 @@ func (p OwnerProjectRequest) ToString() string {
 }
 
 type DatasetRequest struct {
-	OwnerId     int64  `json:"ownerId"`
+	OwnerId     int64   `json:"ownerId"`
-	ProjectId   int64  `json:"projectId"`
+	ProjectId   []int64 `json:"projectId"`
-	DatasetName string `json:"datasetName"`
+	DatasetName string  `json:"datasetName"`
-	StartTime   string `json:"startTime"`
+	StartTime   string  `json:"startTime"`
-	EndTime     string `json:"endTime"`
+	EndTime     string  `json:"endTime"`
 	BasePageList
 }
 
@@ -529,6 +534,20 @@ func (p DatasetRequest) ToString() string {
 	return string(data)
 }
 
+type DatasetItemRequest struct {
+	DatasetId int64 `json:"datasetId"`
+	DataType  int   `json:"dataType"`
+	BasePageList
+}
+
+func (p DatasetItemRequest) ToString() string {
+	data, err := json.Marshal(p)
+	if err != nil {
+		return ""
+	}
+	return string(data)
+}
+
 type ImportDatasetRequest struct {
 	DatasetId   int64  `json:"datasetId"`
 	CategoryId  int    `json:"categoryId"`
@@ -655,3 +674,17 @@ func (p TaskItemRequest) ToString() string {
 	}
 	return string(data)
 }
+
+type ReportRequest struct {
+	ReportId int64 `json:"reportId,omitempty"`
+	TaskId   int64 `json:"taskId,omitempty"`
+	BasePageList
+}
+
+func (p ReportRequest) ToString() string {
+	data, err := json.Marshal(p)
+	if err != nil {
+		return ""
+	}
+	return string(data)
+}

internal/proto/response.go

@@ -121,3 +121,23 @@ type TaskDetail struct {
 	CreateAt        int64  `xorm:"created" json:"createAt"`
 	UpdateAt        int64  `xorm:"updated" json:"updateAt"`
 }
+
+type TaskResultItem struct {
+	FileId              int64   `json:"fileId"`
+	FileName            string  `json:"fileName"`
+	SrcFile             string  `json:"srcFile"`
+	DistFile            string  `json:"distFile"`
+	DiseaseType         int     `json:"diseaseType"`
+	DiseaseTypeName     string  `json:"diseaseTypeName"`
+	DiseaseLevel        int     `json:"diseaseLevel"`
+	DiseaseLevelName    string  `json:"diseaseLevelName"`
+	KPile               string  `json:"KPile"`
+	UpDown              int     `json:"upDown"`
+	LineNum             int     `json:"lineNum"`
+	Length              float64 `json:"length"`
+	Width               float64 `json:"width"`
+	Area                float64 `json:"area"`
+	HorizontalPositions float64 `json:"horizontalPositions"`
+	Memo                string  `json:"memo"`
+	Stat                bool    `json:"stat"`
+}

internal/router/router.go

@@ -1,12 +1,15 @@
 package router
 
 import (
+	"fmt"
 	"git.hpds.cc/Component/logging"
 	ginzap "github.com/gin-contrib/zap"
 	"github.com/gin-gonic/gin"
 	"hpds-iot-web/config"
 	"hpds-iot-web/internal/handler"
 	"hpds-iot-web/internal/middleware"
+	"hpds-iot-web/mq"
+	"net/http"
 	"xorm.io/xorm"
 
 	e "hpds-iot-web/pkg/err"
@@ -48,6 +51,7 @@ func InitRouter(cfg *config.WebConfig, logger *logging.Logger, engine *xorm.Engi
 			project := manage.Group("/project")
 			{
 				project.POST("/list", e.ErrorWrapper(hs.ProjectList))
+				project.POST("/info", e.ErrorWrapper(hs.ProjectInfo))
 				project.POST("/add", e.ErrorWrapper(hs.AddProject))
 				project.POST("/edit", e.ErrorWrapper(hs.EditProject))
 				project.POST("/delete", e.ErrorWrapper(hs.DelProject))
@@ -135,6 +139,7 @@ func InitRouter(cfg *config.WebConfig, logger *logging.Logger, engine *xorm.Engi
 		{
 			file.Use(middleware.JwtAuthMiddleware(logger.Logger))
 			file.POST("/upload", e.ErrorWrapper(hs.UploadFile))
+			file.POST("/list", e.ErrorWrapper(hs.FileList))
 		}
 		system := r.Group("/system")
 		{
@@ -163,6 +168,34 @@ func InitRouter(cfg *config.WebConfig, logger *logging.Logger, engine *xorm.Engi
 			task.Use(middleware.JwtAuthMiddleware(logger.Logger))
 			task.POST("/list", e.ErrorWrapper(hs.TaskList))
 			task.POST("/add", e.ErrorWrapper(hs.AddTask))
+			task.POST("/reRun", e.ErrorWrapper(hs.ReRunTask))
+			task.POST("/info", e.ErrorWrapper(hs.TaskInfo))
+			task.POST("/result", e.ErrorWrapper(hs.TaskResult))
+			task.POST("/log", e.ErrorWrapper(hs.TaskLog))
+			task.GET("/event", func(c *gin.Context) {
+				//us, _ := c.Get("operatorUser")
+				//userInfo := us.(*m.SystemUser)
+
+				mq.AddChannel(fmt.Sprintf("%d", 1))
+				c.Writer.Header().Set("Content-Type", "text/event-stream")
+				c.Writer.Header().Set("Cache-Control", "no-cache")
+				c.Writer.Header().Set("Connection", "keep-alive")
+				w := c.Writer
+				flusher, _ := w.(http.Flusher)
+				closeNotify := c.Request.Context().Done()
+				go func() {
+					<-closeNotify
+					delete(mq.ChannelsMap, fmt.Sprintf("%d", 1))
+					logging.L().Info("SSE close for user = " + fmt.Sprintf("%d", 1))
+					return
+				}()
+				_, _ = fmt.Fprintf(w, "data: %s\n\n", "--ping--")
+				flusher.Flush()
+				for msg := range mq.ChannelsMap[fmt.Sprintf("%d", 1)] {
+					_, _ = fmt.Fprintf(w, "data: %s\n\n", msg)
+					flusher.Flush()
+				}
+			})
 		}
 		disease := r.Group("/disease")
 		{
@@ -183,6 +216,15 @@ func InitRouter(cfg *config.WebConfig, logger *logging.Logger, engine *xorm.Engi
 
 			dataset.POST("/list", e.ErrorWrapper(hs.DatasetList))
 			dataset.POST("/import", e.ErrorWrapper(hs.ImportDataset))
+			dataset.POST("/info", e.ErrorWrapper(hs.DatasetInfo))
+
+		}
+		report := r.Group("/report")
+		{
+			report.Use(middleware.JwtAuthMiddleware(logger.Logger))
+			report.POST("/list", e.ErrorWrapper(hs.ReportList))
+			report.POST("/generate", e.ErrorWrapper(hs.GenerateReport))
+			//report.POST("/view", e.ErrorWrapper(hs.ViewReport))
 		}
 	}
 	return root

internal/service/dataset.go

@@ -16,6 +16,7 @@ type DatasetService interface {
 	GetOwnerProjectList(ctx context.Context, req proto.OwnerProjectRequest) (rsp *proto.BaseResponse, err error)
 	DatasetList(ctx context.Context, req proto.DatasetRequest) (rsp *proto.BaseResponse, err error)
 	ImportDataset(ctx context.Context, req proto.ImportDatasetRequest) (rsp *proto.BaseResponse, err error)
+	DatasetInfo(ctx context.Context, req proto.DatasetItemRequest) (rsp *proto.BaseResponse, err error)
 }
 
 func NewDatasetService(cfg *config.WebConfig, engine *xorm.Engine, logger *logging.Logger) DatasetService {
@@ -86,24 +87,27 @@ func (rp *repo) DatasetList(ctx context.Context, req proto.DatasetRequest) (rsp
 		return rsp, ctx.Err()
 	default:
 		list := make([]model.Dataset, 0)
-		count, err := rp.engine.Where("(? = '' or dataset_name like ?)", req.DatasetName, "%"+req.DatasetName+"%").
+		st := rp.engine.Where("(? = '' or dataset_name like ?)", req.DatasetName, "%"+req.DatasetName+"%").
 			And("(? = '' or create_at >= ?)", req.StartTime, req.StartTime).
 			And("(? = '' or create_at < ?)", req.EndTime, req.EndTime).
 			And("(? = 0 or owner_id = ?)", req.OwnerId, req.OwnerId).
-			And("(? = 0 or project_id = ?)", req.ProjectId, req.ProjectId).
+			And("status = 1")
-			And("status = 1").FindAndCount(&list)
+		if len(req.ProjectId) > 0 {
+			st.In("project_id", req.ProjectId)
+		}
+		count, err := st.Limit(int(req.Size), int(((req.Page)-1)*req.Size)).FindAndCount(&list)
 		if err != nil {
 			goto ReturnPoint
 		}
 		data := make([]proto.DatasetItem, len(list))
 		for k, v := range list {
 			detailList := make([]model.FileManager, 0)
-			datasetCount, err := rp.engine.Where("dataset_id = ?", v.DatasetId).FindAndCount(&detailList)
+			datasetCount, err := rp.engine.Where("dataset_id = ?", v.DatasetId).And("data_type=1").FindAndCount(&detailList)
 			if err != nil {
 				goto ReturnPoint
 			}
 			fm := new(model.FileManager)
-			datasetSize, err := rp.engine.Where("dataset_id = ?", v.DatasetId).SumInt(fm, "file_size")
+			datasetSize, err := rp.engine.Where("dataset_id = ?", v.DatasetId).And("data_type=1").SumInt(fm, "file_size")
 			if err != nil {
 				goto ReturnPoint
 			}
@@ -181,3 +185,42 @@ ReturnPoint:
 	}
 	return rsp, err
 }
+
+func (rp *repo) DatasetInfo(ctx context.Context, req proto.DatasetItemRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		item := new(model.Dataset)
+		var b bool
+		b, err = rp.engine.ID(req.DatasetId).Get(item)
+		if err != nil {
+			goto ReturnPoint
+		}
+		if !b {
+			err = fmt.Errorf("未能找到对应的数据集")
+			goto ReturnPoint
+		}
+
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp.Err = err
+		rsp.Data = item
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}

internal/service/fileManage.go

@@ -22,7 +22,8 @@ import (
 
 type FileService interface {
 	UploadFile(ctx context.Context, req proto.UploadFileRequest) (rsp *proto.BaseResponse, err error)
-	UploadFileToMinIo(ctx context.Context, srcFile *multipart.FileHeader, scene string, datasetId, creator int64) (data *model.FileManager, err error)
+	UploadFileToMinIo(ctx context.Context, srcFile *multipart.FileHeader, scene string, datasetId, creator int64, dataType int) (data *model.FileManager, err error)
+	FileList(ctx context.Context, req proto.DatasetItemRequest) (rsp *proto.BaseResponse, err error)
 }
 
 func NewFileService(cfg *config.WebConfig, engine *xorm.Engine, logger *logging.Logger) FileService {
@@ -47,7 +48,7 @@ func (rp *repo) UploadFile(ctx context.Context, req proto.UploadFileRequest) (rs
 		list := make([]*model.FileManager, len(req.Files))
 		fileUrl := make([]string, len(req.Files))
 		for k := range req.Files {
-			fileItem, err := rp.UploadFileToMinIo(ctx, req.Files[k], req.Scene, req.DatasetId, req.Creator)
+			fileItem, err := rp.UploadFileToMinIo(ctx, req.Files[k], req.Scene, req.DatasetId, req.Creator, req.DataType)
 			if err != nil {
 				goto ReturnPoint
 			}
@@ -76,7 +77,7 @@ ReturnPoint:
 	return rsp, err
 }
 
-func (rp *repo) UploadFileToMinIo(ctx context.Context, srcFile *multipart.FileHeader, scene string, datasetId, creator int64) (data *model.FileManager, err error) {
+func (rp *repo) UploadFileToMinIo(ctx context.Context, srcFile *multipart.FileHeader, scene string, datasetId, creator int64, dataType int) (data *model.FileManager, err error) {
 	file, err := srcFile.Open()
 	defer func(file multipart.File) {
 		_ = file.Close()
@@ -126,9 +127,47 @@ func (rp *repo) UploadFileToMinIo(ctx context.Context, srcFile *multipart.FileHe
 	data.Scene = scene
 	data.AccessUrl = accessUrl
 	data.FileSize = srcFile.Size
+	data.DataType = dataType
+	data.DatasetId = datasetId
 	data.Creator = creator
 	data.FileMd5 = hex.EncodeToString(md5hash.Sum(nil))
 	data.CreateAt = time.Now().Unix()
 	data.UpdateAt = time.Now().Unix()
 	return data, nil
 }
+
+func (rp *repo) FileList(ctx context.Context, req proto.DatasetItemRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		data := make([]model.FileManager, 0)
+		count, err := rp.engine.Where("(? = 0 or dataset_id = ?)", req.DatasetId, req.DatasetId).
+			And("(? = 0 or data_type = ?)", req.DataType, req.DataType).
+			Limit(int(req.Size), int(((req.Page)-1)*req.Size)).
+			FindAndCount(&data)
+		if err != nil {
+			goto ReturnPoint
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp = FillPaging(count, req.Page, req.Size, data, rsp)
+		rsp.Err = err
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}

internal/service/manage.go

@@ -24,6 +24,7 @@ type ManageService interface {
 
 	ProjectList(ctx context.Context, req proto.ProjectRequest) (rsp *proto.BaseResponse, err error)
 	GetLngLat(ctx context.Context, mileagePile string) (lng, lat float64, err error)
+	ProjectInfo(ctx context.Context, req proto.ProjectItemRequest) (rsp *proto.BaseResponse, err error)
 	AddProject(ctx context.Context, req proto.ProjectItemRequest) (rsp *proto.BaseResponse, err error)
 	EditProject(ctx context.Context, req proto.ProjectItemRequest) (rsp *proto.BaseResponse, err error)
 	DelProject(ctx context.Context, req proto.ProjectItemRequest) (rsp *proto.BaseResponse, err error)
@@ -321,6 +322,44 @@ ReturnPoint:
 	return rsp, err
 }
 
+func (rp *repo) ProjectInfo(ctx context.Context, req proto.ProjectItemRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		var h bool
+		item := new(model.Project)
+		h, err = rp.engine.ID(req.ProjectId).Get(item)
+		if err != nil {
+			goto ReturnPoint
+		}
+		if !h {
+			err = fmt.Errorf("未能找到对应的项目")
+			goto ReturnPoint
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp.Data = item
+		rsp.Err = err
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}
+
 func (rp *repo) AddProject(ctx context.Context, req proto.ProjectItemRequest) (rsp *proto.BaseResponse, err error) {
 	rsp = new(proto.BaseResponse)
 	select {
@@ -378,6 +417,7 @@ ReturnPoint:
 	}
 	return rsp, err
 }
+
 func (rp *repo) GetLngLat(ctx context.Context, mileagePile string) (lng, lat float64, err error) {
 	mine := minedata.NewMineData(rp.AppConfig.MineData.AccessKey)
 	//正里程桩

internal/service/report.go

@@ -0,0 +1,187 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"git.hpds.cc/Component/logging"
+	"hpds-iot-web/internal/proto"
+	"hpds-iot-web/model"
+	"net/http"
+	"xorm.io/xorm"
+
+	"github.com/360EntSecGroup-Skylar/excelize"
+)
+
+type ReportService interface {
+	ReportList(ctx context.Context, req proto.ReportRequest) (rsp *proto.BaseResponse, err error)
+	GenerateReport(ctx context.Context, req proto.ReportRequest) (rsp *excelize.File, err error)
+}
+
+func NewReportService(engine *xorm.Engine, logger *logging.Logger) ReportService {
+	return &repo{
+		engine: engine,
+		logger: logger,
+	}
+}
+
+func (rp *repo) ReportList(ctx context.Context, req proto.ReportRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		data := make([]model.Report, 0)
+		count, err := rp.engine.
+			Where("(? = 0 or m.report_id = ?)", req.ReportId, req.ReportId).
+			And("t.status > 0").Limit(int(req.Size), int(((req.Page)-1)*req.Size)).
+			FindAndCount(&data)
+		if err != nil {
+			goto ReturnPoint
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp = FillPaging(count, req.Page, req.Size, data, rsp)
+		rsp.Err = err
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}
+
+func (rp *repo) GenerateReport(ctx context.Context, req proto.ReportRequest) (rsp *excelize.File, err error) {
+	//rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		//rsp.Code = http.StatusInternalServerError
+		//rsp.Status = http.StatusText(http.StatusInternalServerError)
+		//rsp.Message = "超时/取消"
+		//rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		//data := make([]model.Report, 0)
+		//_, err = rp.engine.
+		//	Where("(? = 0 or m.report_id = ?)", req.ReportId, req.ReportId).
+		//	And("t.status > 0").Limit(int(req.Size), int(((req.Page)-1)*req.Size)).
+		//	FindAndCount(&data)
+		//if err != nil {
+		//	goto ReturnPoint
+		//}
+		list := make([]proto.TaskResultItem, 2)
+
+		list[0] = proto.TaskResultItem{
+			FileId:              147,
+			FileName:            "Y016   2.844 I 1 06272008 Color.jpg",
+			SrcFile:             "http://127.0.0.1:9000/jky-data/road/test_dataset/Y016   2.844 I 1 06272008 Color.jpg",
+			DistFile:            "http://127.0.0.1:9000/jky-data/road/test_dataset/Y016   2.844 I 1 06272008 Color.jpg",
+			DiseaseType:         4,
+			DiseaseLevel:        2,
+			KPile:               "K229+778",
+			UpDown:              1,
+			LineNum:             2,
+			Length:              0.5,
+			Width:               1,
+			Area:                0.5,
+			HorizontalPositions: 0.5,
+			Memo:                "有裂缝",
+		}
+		list[1] = proto.TaskResultItem{
+			FileId:              149,
+			FileName:            "C279   0.132 I 1 06272008 Color.jpg",
+			SrcFile:             "http://127.0.0.1:9000/jky-data/road/test_dataset/cba/C279   0.132 I 1 06272008 Color.jpg",
+			DistFile:            "http://127.0.0.1:9000/jky-data/road/test_dataset/cba/C279   0.132 I 1 06272008 Color.jpg",
+			DiseaseType:         0,
+			DiseaseLevel:        0,
+			KPile:               "",
+			UpDown:              0,
+			LineNum:             0,
+			Length:              0,
+			Width:               0,
+			Area:                0,
+			HorizontalPositions: 0,
+		}
+		disTypeList := make([]model.DiseaseType, 0)
+		err = model.DB.Find(&disTypeList)
+		if err != nil {
+			goto ReturnPoint
+		}
+		diseaseTypeList := make(map[int64]string)
+		for _, v := range disTypeList {
+			diseaseTypeList[v.TypeId] = v.TypeName
+		}
+		diseaseLevelList := make(map[int]string)
+		diseaseLevelList[1] = "一级"
+		diseaseLevelList[2] = "二级"
+		diseaseLevelList[3] = "三级"
+
+		xlsx := excelize.NewFile()
+		xlsx.MergeCell("Sheet1", "A1", "L1")
+		xlsx.MergeCell("Sheet1", "A2", "L2")
+		xlsx.MergeCell("Sheet1", "A3", "L3")
+		xlsx.MergeCell("Sheet1", "A4", "L4")
+		xlsx.SetCellValue("Sheet1", "A4", "病害明细")
+		//序号	路线编码	方向	桩号	车道	病害名称	程度	长度(m)	宽度(m)	面积(㎡)	横向位置	备注
+
+		xlsx.SetCellValue("Sheet1", "A5", "序号")
+		xlsx.SetCellValue("Sheet1", "B5", "路线编码")
+		xlsx.SetCellValue("Sheet1", "C5", "方向")
+		xlsx.SetCellValue("Sheet1", "D5", "桩号")
+		xlsx.SetCellValue("Sheet1", "E5", "车道")
+		xlsx.SetCellValue("Sheet1", "F5", "病害名称")
+		xlsx.SetCellValue("Sheet1", "G5", "程度")
+		xlsx.SetCellValue("Sheet1", "H5", "长度(m)")
+		xlsx.SetCellValue("Sheet1", "I5", "宽度(m)")
+		xlsx.SetCellValue("Sheet1", "J5", "面积(㎡)")
+		xlsx.SetCellValue("Sheet1", "K5", "横向位置")
+		xlsx.SetCellValue("Sheet1", "L5", "备注")
+		for k, v := range list {
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("A%d", k+6), k+1)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("B%d", k+6), v.KPile)
+			switch v.UpDown {
+			case 1:
+				xlsx.SetCellValue("Sheet1", fmt.Sprintf("C%d", k+6), "上行")
+			case 2:
+				xlsx.SetCellValue("Sheet1", fmt.Sprintf("C%d", k+6), "下行")
+			}
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("D%d", k+6), v.KPile)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("E%d", k+6), v.LineNum)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("F%d", k+6), diseaseTypeList[int64(v.DiseaseType)])
+
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("G%d", k+6), diseaseLevelList[v.DiseaseLevel])
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("H%d", k+6), v.Length)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("I%d", k+6), v.Width)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("J%d", k+6), v.Area)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("K%d", k+6), v.HorizontalPositions)
+			xlsx.SetCellValue("Sheet1", fmt.Sprintf("K%d", k+6), v.Memo)
+
+		}
+		//rsp = xlsx
+		//rsp.Code = http.StatusOK
+		//rsp.Status = http.StatusText(http.StatusOK)
+		//rsp.Message = "成功"
+		//rsp.Data = xlsx
+		//rsp.Err = err
+		return xlsx, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp = nil
+		//rsp.Code = http.StatusInternalServerError
+		//rsp.Status = http.StatusText(http.StatusInternalServerError)
+		//rsp.Err = err
+		//rsp.Message = "失败"
+	}
+	return rsp, err
+}

internal/service/system.go

@@ -411,7 +411,7 @@ func (rp *repo) NodeState(ctx context.Context, req proto.NodeInfoRequest) (rsp *
 		return rsp, ctx.Err()
 	default:
 		list := make([]model.NodeState, 0)
-		err = rp.engine.Where("node_name = ?", req.NodeGuid).
+		err = rp.engine.Where("node_guid = ?", req.NodeGuid).
 			And(" uptime > UNIX_TIMESTAMP(DATE_ADD(NOW(),INTERVAL -24 HOUR))").
 			Find(&list)
 		if err != nil {

internal/service/task.go

@@ -1,7 +1,9 @@
 package service
 
 import (
+	"bytes"
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"git.hpds.cc/Component/logging"
@@ -9,15 +11,21 @@ import (
 	"hpds-iot-web/internal/proto"
 	"hpds-iot-web/model"
 	"hpds-iot-web/mq"
+	"image"
 	"net/http"
+	"strconv"
 	"time"
 	"xorm.io/xorm"
 )
 
 type TaskService interface {
 	TaskList(ctx context.Context, req proto.TaskRequest) (rsp *proto.BaseResponse, err error)
+	TaskInfo(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error)
 	AddTask(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error)
+	ReRunTask(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error)
 	//EditTask(ctx context.Context, req proto.ModelItemRequest) (rsp *proto.BaseResponse, err error)
+	TaskResult(ctx context.Context, req proto.ReportRequest) (rsp *proto.BaseResponse, err error)
+	TaskLog(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error)
 }
 
 func NewTaskService(engine *xorm.Engine, logger *logging.Logger) TaskService {
@@ -48,6 +56,7 @@ func (rp *repo) TaskList(ctx context.Context, req proto.TaskRequest) (rsp *proto
 			And("t.start_time >= unix_timestamp(?)", req.StartTime).
 			And("? = 0 or t.start_time <= unix_timestamp(?)", req.FinishTime, req.FinishTime).
 			And("t.status > 0").Limit(int(req.Size), int(((req.Page)-1)*req.Size)).
+			Desc("start_time").
 			FindAndCount(&data)
 		if err != nil {
 			goto ReturnPoint
@@ -92,7 +101,7 @@ func (rp *repo) AddTask(ctx context.Context, req proto.TaskItemRequest) (rsp *pr
 		}
 
 		ds := new(model.Dataset)
-		h, err = rp.engine.ID(req.ModelId).Get(ds)
+		h, err = rp.engine.ID(req.DatasetArr).Get(ds)
 		if err != nil {
 			goto ReturnPoint
 		}
@@ -100,6 +109,18 @@ func (rp *repo) AddTask(ctx context.Context, req proto.TaskItemRequest) (rsp *pr
 			err = fmt.Errorf("未能找到对应的数据集")
 			goto ReturnPoint
 		}
+		var node *model.Node
+		if req.NodeId > 0 {
+			node = new(model.Node)
+			h, err = rp.engine.ID(req.NodeId).Get(node)
+			if err != nil {
+				goto ReturnPoint
+			}
+			if !h {
+				err = fmt.Errorf("未能找到对应的节点")
+				goto ReturnPoint
+			}
+		}
 
 		item := &model.Task{
 			ModelId:         req.ModelId,
@@ -143,6 +164,9 @@ func (rp *repo) AddTask(ctx context.Context, req proto.TaskItemRequest) (rsp *pr
 		payload["modelVersion"] = m.ModelVersion
 		payload["modelCommand"] = m.ModelCommand
 		payload["nodeId"] = item.NodeId
+		if item.NodeId > 0 && node != nil {
+			payload["nodeGuid"] = node.NodeGuid
+		}
 		payload["inPath"] = m.InPath
 		payload["outPath"] = m.OutPath
 		payload["httpUrl"] = m.HttpUrl
@@ -153,6 +177,11 @@ func (rp *repo) AddTask(ctx context.Context, req proto.TaskItemRequest) (rsp *pr
 		payload["subDataTag"] = item.SubDataTag
 		payload["workflow"] = m.Workflow
 
+		issue := new(model.IssueModel)
+		h, _ = model.DB.Where("model_id=? and node_id =?", req.ModelId, item.NodeId).Get(issue)
+		if h {
+			payload["issueResult"] = issue.IssueResult
+		}
 		mqClient := mq.GetMqClient("task-request", 1)
 		mqPayload := &mq.InstructionReq{
 			Command: mq.TaskAdd,
@@ -176,3 +205,473 @@ ReturnPoint:
 	}
 	return rsp, err
 }
+
+func (rp *repo) ReRunTask(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		item := new(model.Task)
+		var h bool
+		h, err = rp.engine.ID(req.TaskId).Get(item)
+		if err != nil {
+			goto ReturnPoint
+		}
+		if !h {
+			err = fmt.Errorf("未能找到任务")
+			goto ReturnPoint
+		}
+		m := new(model.Model)
+		h, err = rp.engine.ID(item.ModelId).Get(m)
+		if err != nil {
+			goto ReturnPoint
+		}
+		if !h {
+			err = fmt.Errorf("未能找到对应的模型")
+			goto ReturnPoint
+		}
+		ds := new(model.Dataset)
+		h, err = rp.engine.ID(item.DatasetArr).Get(ds)
+		if err != nil {
+			goto ReturnPoint
+		}
+		if !h {
+			err = fmt.Errorf("未能找到对应的数据集")
+			goto ReturnPoint
+		}
+		var node *model.Node
+		if item.NodeId > 0 {
+			node = new(model.Node)
+			h, err = rp.engine.ID(item.NodeId).Get(node)
+			if err != nil {
+				goto ReturnPoint
+			}
+			if !h {
+				err = fmt.Errorf("未能找到对应的节点")
+				goto ReturnPoint
+			}
+		}
+		payload := make(map[string]interface{})
+		payload["taskId"] = item.TaskId
+		payload["modelId"] = item.ModelId
+		payload["modelVersion"] = m.ModelVersion
+		payload["modelCommand"] = m.ModelCommand
+		payload["nodeId"] = item.NodeId
+		if item.NodeId > 0 && node != nil {
+			payload["nodeGuid"] = node.NodeGuid
+		}
+		payload["inPath"] = m.InPath
+		payload["outPath"] = m.OutPath
+		payload["httpUrl"] = m.HttpUrl
+		payload["datasetArr"] = item.DatasetArr
+		payload["datasetPath"] = ds.StoreName
+		payload["datasetName"] = ds.DatasetName
+		payload["subDataset"] = item.SubDataset
+		payload["subDataTag"] = item.SubDataTag
+		payload["workflow"] = m.Workflow
+
+		issue := new(model.IssueModel)
+		h, _ = model.DB.Where("model_id=? and node_id =?", req.ModelId, item.NodeId).Get(issue)
+		if h {
+			payload["issueResult"] = issue.IssueResult
+		}
+		mqClient := mq.GetMqClient("task-request", 1)
+		mqPayload := &mq.InstructionReq{
+			Command: mq.TaskAdd,
+			Payload: payload,
+		}
+		pData, _ := json.Marshal(mqPayload)
+		err = mq.GenerateAndSendData(mqClient.EndPoint.(hpds_node.AccessPoint), pData, rp.logger)
+		if err != nil {
+			goto ReturnPoint
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "重新任务成功"
+		rsp.Err = ctx.Err()
+		rsp.Data = item
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}
+
+func (rp *repo) TaskInfo(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		item := new(model.Task)
+		var b bool
+		b, err = rp.engine.ID(req.TaskId).Get(item)
+		if err != nil {
+			goto ReturnPoint
+		}
+		if !b {
+			err = fmt.Errorf("未能找到对应的任务")
+			goto ReturnPoint
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp.Err = err
+		rsp.Data = item
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}
+
+func (rp *repo) TaskResult(ctx context.Context, req proto.ReportRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		taskResultList := make([]model.TaskResult, 0)
+		err = rp.engine.Where("task_id = ?", req.TaskId).
+			Limit(int(req.Size), int(((req.Page)-1)*req.Size)).
+			Find(&taskResultList)
+		if err != nil {
+			err = fmt.Errorf("未能找到对应的结果")
+			goto ReturnPoint
+		}
+		list := make([]proto.TaskResultItem, 0)
+		for _, v := range taskResultList {
+			var h bool
+			file := new(model.FileManager)
+			h, err = rp.engine.ID(v.FileId).Get(file)
+			if err != nil || !h {
+				continue
+			}
+			md := new(model.Model)
+			h, err = rp.engine.ID(v.ModelId).Get(md)
+			if err != nil || !h {
+				continue
+			}
+			var (
+				mr               mq.ModelResult
+				mrList           []string
+				fileDiscern      string
+				memo             string
+				diseaseType      int64
+				diseaseLevel     int
+				length           float64
+				area             float64
+				width            float64
+				diseaseTypeName  string
+				diseaseLevelName string
+			)
+			if len(v.Result) > 0 && v.Result[0] == '[' {
+				mrList = make([]string, 0)
+				if err := json.Unmarshal([]byte(v.Result), &mrList); err != nil {
+					continue
+				}
+				for _, str := range mrList {
+					if err := json.Unmarshal([]byte(str), &mr); err != nil {
+						continue
+					}
+					switch mr.Code {
+					case 0: //轻量化模型返回
+						lr := new(mq.LightweightResult)
+						if err := json.Unmarshal([]byte(v.Result), lr); err != nil {
+							continue
+						}
+						//for _, val := range lrList {
+						if lr.Crack || lr.Pothole {
+							if lr.Crack {
+								memo = "检测到裂缝"
+							} else {
+								memo = "检测到坑洼"
+							}
+							fileDiscern = lr.ImgDiscern
+							diseaseLevel = 3
+							diseaseLevelName = "重度"
+							switch md.BizType {
+							case 2:
+								diseaseType = 8
+								diseaseTypeName = "结构裂缝"
+							case 3:
+								diseaseType = 15
+								diseaseTypeName = "衬砌裂缝"
+							default:
+								diseaseType = 4
+								diseaseTypeName = "横向裂缝"
+							}
+						}
+						fn, _ := base64.StdEncoding.DecodeString(fileDiscern)
+						buff := bytes.NewBuffer(fn)
+						_, imgType, _ := image.Decode(buff)
+						fileDiscern = fmt.Sprintf("data:image/%s;base64,%s", imgType, fileDiscern)
+						item := proto.TaskResultItem{
+							FileId:              v.FileId,
+							FileName:            v.SrcPath,
+							SrcFile:             file.AccessUrl,
+							DistFile:            fileDiscern,
+							DiseaseType:         int(diseaseType),
+							DiseaseTypeName:     diseaseTypeName,
+							DiseaseLevel:        diseaseLevel,
+							DiseaseLevelName:    diseaseLevelName,
+							KPile:               "",
+							UpDown:              0,
+							LineNum:             0,
+							Length:              length,
+							Width:               width,
+							Area:                area,
+							HorizontalPositions: 0,
+							Memo:                memo,
+							Stat:                false,
+						}
+						list = append(list, item)
+						//}
+					case 2000:
+						ir := new(mq.InsigmaResult)
+						if err := json.Unmarshal([]byte(str), &ir); err != nil {
+							continue
+						}
+						fileDiscern = ir.Image
+						fn, _ := base64.StdEncoding.DecodeString(fileDiscern)
+						buff := bytes.NewBuffer(fn)
+						_, imgType, _ := image.Decode(buff)
+						fileDiscern = fmt.Sprintf("data:image/%s;base64,%s", imgType, fileDiscern)
+						item := proto.TaskResultItem{
+							FileId:              v.FileId,
+							FileName:            v.SrcPath,
+							SrcFile:             file.AccessUrl,
+							DistFile:            fileDiscern,
+							DiseaseType:         int(diseaseType),
+							DiseaseTypeName:     diseaseTypeName,
+							DiseaseLevel:        diseaseLevel,
+							DiseaseLevelName:    diseaseLevelName,
+							KPile:               "",
+							UpDown:              0,
+							LineNum:             0,
+							Length:              length,
+							Width:               width,
+							Area:                area,
+							HorizontalPositions: 0,
+							Memo:                memo,
+							Stat:                false,
+						}
+						list = append(list, item)
+
+					case 2001:
+						ir := new(mq.InsigmaResult)
+						if err := json.Unmarshal([]byte(str), &ir); err != nil {
+							continue
+						}
+						fileDiscern = ir.Image
+						for _, value := range ir.Diseases {
+							diseaseType = model.GetDiseaseType(value.Type, md.BizType)
+							if len(value.Param.MaxWidth) > 0 && width == 0 {
+								width, _ = strconv.ParseFloat(value.Param.MaxWidth, 64)
+							}
+							length = value.Param.Length
+							area = value.Param.Area
+							diseaseLevelName = value.Level
+							diseaseTypeName = value.Type
+							switch value.Level {
+							case "重度":
+								diseaseLevel = 3
+							case "中度":
+								diseaseLevel = 2
+							case "轻度":
+								diseaseLevel = 1
+							}
+							memo += fmt.Sprintf("发现[%s],等级[%s],长度[%f],最大宽度[%s],面积[%f];\n", value.Type, value.Level, value.Param.Length, value.Param.MaxWidth, value.Param.Area)
+						}
+						fn, _ := base64.StdEncoding.DecodeString(fileDiscern)
+						buff := bytes.NewBuffer(fn)
+						_, imgType, _ := image.Decode(buff)
+						fileDiscern = fmt.Sprintf("data:image/%s;base64,%s", imgType, fileDiscern)
+						item := proto.TaskResultItem{
+							FileId:              v.FileId,
+							FileName:            v.SrcPath,
+							SrcFile:             file.AccessUrl,
+							DistFile:            fileDiscern,
+							DiseaseType:         int(diseaseType),
+							DiseaseTypeName:     diseaseTypeName,
+							DiseaseLevel:        diseaseLevel,
+							DiseaseLevelName:    diseaseLevelName,
+							KPile:               "",
+							UpDown:              0,
+							LineNum:             0,
+							Length:              length,
+							Width:               width,
+							Area:                area,
+							HorizontalPositions: 0,
+							Memo:                memo,
+							Stat:                false,
+						}
+						list = append(list, item)
+					}
+				}
+
+			} else {
+				if err := json.Unmarshal([]byte(v.Result), &mr); err != nil {
+					continue
+				}
+				switch mr.Code {
+				case 0: //轻量化模型返回
+					lr := new(mq.LightweightResult)
+					if err := json.Unmarshal([]byte(v.Result), &lr); err != nil {
+						continue
+					}
+					if lr.Crack || lr.Pothole {
+						if lr.Crack {
+							memo = "检测到裂缝"
+						} else {
+							memo = "检测到坑洼"
+						}
+						fileDiscern = lr.ImgDiscern
+						diseaseLevel = 3
+						diseaseLevelName = "重度"
+						switch md.BizType {
+						case 2:
+							diseaseType = 8
+							diseaseTypeName = "结构裂缝"
+						case 3:
+							diseaseType = 15
+							diseaseTypeName = "衬砌裂缝"
+						default:
+							diseaseType = 4
+							diseaseTypeName = "横向裂缝"
+						}
+					}
+					//
+				case 2000: //网新返回没有病害
+					ir := new(mq.InsigmaResult)
+					if err := json.Unmarshal([]byte(v.Result), &ir); err != nil {
+						continue
+					}
+					fileDiscern = ir.Image
+				case 2001: //网新返回有病害
+					ir := new(mq.InsigmaResult)
+					if err := json.Unmarshal([]byte(v.Result), &ir); err != nil {
+						continue
+					}
+					fileDiscern = ir.Image
+					for _, val := range ir.Diseases {
+						diseaseType = model.GetDiseaseType(val.Type, md.BizType)
+						if len(val.Param.MaxWidth) > 0 && width == 0 {
+							width, _ = strconv.ParseFloat(val.Param.MaxWidth, 64)
+						}
+						length = val.Param.Length
+						area = val.Param.Area
+						diseaseLevelName = val.Level
+						diseaseTypeName = val.Type
+						switch val.Level {
+						case "重度":
+							diseaseLevel = 3
+						case "中度":
+							diseaseLevel = 2
+						case "轻度":
+							diseaseLevel = 1
+						}
+						memo += fmt.Sprintf("发现[%s],等级[%s],长度[%f],最大宽度[%s],面积[%f];\n", val.Type, val.Level, val.Param.Length, val.Param.MaxWidth, val.Param.Area)
+					}
+				}
+				fn, _ := base64.StdEncoding.DecodeString(fileDiscern)
+				buff := bytes.NewBuffer(fn)
+				_, imgType, _ := image.Decode(buff)
+				fileDiscern = fmt.Sprintf("data:image/%s;base64,%s", imgType, fileDiscern)
+				item := proto.TaskResultItem{
+					FileId:              v.FileId,
+					FileName:            v.SrcPath,
+					SrcFile:             file.AccessUrl,
+					DistFile:            fileDiscern,
+					DiseaseType:         int(diseaseType),
+					DiseaseTypeName:     diseaseTypeName,
+					DiseaseLevel:        diseaseLevel,
+					DiseaseLevelName:    diseaseLevelName,
+					KPile:               "",
+					UpDown:              0,
+					LineNum:             0,
+					Length:              length,
+					Width:               width,
+					Area:                area,
+					HorizontalPositions: 0,
+					Memo:                memo,
+					Stat:                false,
+				}
+				list = append(list, item)
+			}
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp.Err = err
+		rsp.Data = list
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}
+
+func (rp *repo) TaskLog(ctx context.Context, req proto.TaskItemRequest) (rsp *proto.BaseResponse, err error) {
+	rsp = new(proto.BaseResponse)
+	select {
+	case <-ctx.Done():
+		err = fmt.Errorf("超时/取消")
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Message = "超时/取消"
+		rsp.Err = ctx.Err()
+		return rsp, ctx.Err()
+	default:
+		list := make([]model.TaskLog, 0)
+		err = rp.engine.Where("task_id = ?", req.TaskId).Find(&list)
+		if err != nil {
+			goto ReturnPoint
+		}
+		rsp.Code = http.StatusOK
+		rsp.Status = http.StatusText(http.StatusOK)
+		rsp.Message = "成功"
+		rsp.Err = err
+		rsp.Data = list
+		return rsp, err
+	}
+ReturnPoint:
+	if err != nil {
+		rsp.Code = http.StatusInternalServerError
+		rsp.Status = http.StatusText(http.StatusInternalServerError)
+		rsp.Err = err
+		rsp.Message = "失败"
+	}
+	return rsp, err
+}

model/TaskLog.go

@@ -0,0 +1,14 @@
+package model
+
+type TaskLog struct {
+	TaskLogId int64  `xorm:"not null pk autoincr BIGINT(11)" json:"taskLogId"`
+	TaskId    int64  `xorm:"INT(11) index" json:"taskId"`
+	NodeId    int64  `xorm:"INT(11) index" json:"nodeId"`
+	Content   string `xorm:"LANGTEXT" json:"content"`
+	CreateAt  int64  `xorm:"created" json:"createAt"`
+	UpdateAt  int64  `xorm:"updated" json:"updateAt"`
+}
+
+func InsertLog(taskLog *TaskLog) {
+	_, _ = DB.Insert(taskLog)
+}

model/diseaseType.go

@@ -9,3 +9,15 @@ type DiseaseType struct {
 	CreateAt   int64  `xorm:"created" json:"createAt"`
 	UpdateAt   int64  `xorm:"updated" json:"updateAt"`
 }
+
+func GetDiseaseType(name string, categoryId int) int64 {
+	item := new(DiseaseType)
+	h, err := DB.Where("type_name like ?", "%"+name+"%").
+		And("category_id = ?", categoryId).
+		And("status = 1").Get(item)
+	if err != nil || !h {
+		return 0
+	}
+	return item.TypeId
+
+}

model/file.go

@@ -5,6 +5,7 @@ type FileManager struct {
 	FileName  string `xorm:"VARCHAR(200)" json:"fileName"`               //文件名
 	AccessUrl string `xorm:"VARCHAR(400)" json:"url"`                    //访问路径
 	Scene     string `xorm:"VARCHAR(40)" json:"scene"`                   //应用场景，0 : 其他, 1: 道路; 2: 桥梁; 3:隧道; 4: 边坡
+	DataType  int    `xorm:"TINYINT index default 0" json:"dataType"`    //数据类型，1:数据集;2:病害库;3:应用;0:其他
 	DatasetId int64  `xorm:"INT(11) index default 0" json:"datasetId"`   //数据集
 	FileSize  int64  `xorm:"BIGINT" json:"fileSize"`                     //文件大小
 	FileMd5   string `xorm:"VARCHAR(64)" json:"fileMd5"`                 //文件MD5

model/index.go

@@ -59,6 +59,7 @@ func New(driveName, dsn string, showSql bool) {
 		&SystemUser{},
 		&SystemUserRole{},
 		&Task{},
+		&TaskLog{},
 		&TaskResult{},
 	)
 	if err != nil {

model/nodeState.go

@@ -3,7 +3,8 @@ package model
 // NodeState 节点状态信息
 type NodeState struct {
 	Uptime         uint64  `xorm:"BIGINT pk" json:"uptime,omitempty"`
-	NodeName       string  `xorm:"varchar(100) pk" json:"nodeName"`
+	NodeGuid       string  `xorm:"varchar(100) pk" json:"nodeGuid,omitempty"`
+	NodeName       string  `xorm:"varchar(100)" json:"nodeName"`
 	CPU            float64 `xorm:"DECIMAL(18,4)" json:"cpu,omitempty"`
 	MemUsed        uint64  `xorm:"BIGINT" json:"memUsed,omitempty"`
 	SwapUsed       uint64  `xorm:"BIGINT" json:"swapUsed,omitempty"`

model/report.go

@@ -0,0 +1,9 @@
+package model
+
+type Report struct {
+	ReportId int64 `xorm:"not null pk autoincr INT(11)" json:"reportId"`
+	TaskId   int64 `xorm:"INT(11) index" json:"taskId"`
+	Status   int   `xorm:"TINYINT  default 1" json:"status"`
+	CreateAt int64 `xorm:"created" json:"createAt"`
+	UpdateAt int64 `xorm:"updated" json:"updateAt"`
+}

model/reportDetail.go

@@ -0,0 +1,13 @@
+package model
+
+type ReportDetail struct {
+	DetailId   int64  `xorm:"not null pk autoincr INT(11)" json:"detailId"`
+	ReportId   int64  `xorm:"INT(11) index" json:"reportId"`
+	FileId     int64  `xorm:"INT(11) index" json:"fileId"`
+	FieldName  string `xorm:"VARCHAR(100)" json:"fieldName"`
+	FieldValue string `xorm:"LONGTEXT" json:"fieldValue"`
+	Status     int    `xorm:"TINYINT default 1" json:"status"`
+	CreateAt   int64  `xorm:"created" json:"createAt"`
+	UpdateAt   int64  `xorm:"updated" json:"updateAt"`
+	DeleteAt   int64  `xorm:"deleted" json:"deleteAt"`
+}

model/task.go

@@ -12,6 +12,10 @@ type Task struct {
 	AppointmentTime string `xorm:"VARCHAR(30)" json:"appointmentTime"`
 	StartTime       int64  `xorm:"BIGINT" json:"startTime"`
 	FinishTime      int64  `xorm:"BIGINT" json:"finishTime"`
+	TotalCount      int64  `xorm:"INT" json:"totalCount"`
+	FailingCount    int64  `xorm:"INT" json:"failingCount"`
+	CompletedCount  int64  `xorm:"INT" json:"completedCount"`
+	UnfinishedCount int64  `xorm:"INT" json:"unfinishedCount"`
 	Status          int    `xorm:"not null SMALLINT default 0" json:"status"` // 1:等待执行; 2:执行中; 3:执行完成; 4:任务分配失败; 5:任务执行失败
 	CreateAt        int64  `xorm:"created" json:"createAt"`
 	UpdateAt        int64  `xorm:"updated" json:"updateAt"`

model/taskResult.go

@@ -11,5 +11,6 @@ type TaskResult struct {
 	SubDataset string `xorm:"varchar(200)" json:"subDataset"`
 	DatasetId  int64  `xorm:"INT(11)" json:"datasetId"`
 	SrcPath    string `xorm:"varchar(500)" json:"srcPath"`
-	Result     string `xorm:"TEXT" json:"result"`
+	Result     string `xorm:"LONGTEXT" json:"result"`
+	FileId     int64  `xorm:"BIGINT" json:"fileId"`
 }

mq/index.go

@@ -1,8 +1,10 @@
 package mq
 
 import (
+	"encoding/json"
 	"fmt"
 	"git.hpds.cc/Component/logging"
+	"git.hpds.cc/Component/network/frame"
 	"go.uber.org/zap"
 	"hpds-iot-web/config"
 	"os"
@@ -20,6 +22,23 @@ type HpdsMqNode struct {
 	EndPoint interface{}
 }
 
+var ifChannelsMapInit = false
+
+var ChannelsMap = map[string]chan string{}
+
+func initChannelsMap() {
+	ChannelsMap = make(map[string]chan string)
+}
+func AddChannel(userId string) {
+	if !ifChannelsMapInit {
+		initChannelsMap()
+		ifChannelsMapInit = true
+	}
+	var newChannel = make(chan string)
+	ChannelsMap[userId] = newChannel
+	logging.L().Info("Build SSE connection for user = " + userId)
+}
+
 func must(logger *logging.Logger, err error) {
 	if err != nil {
 		if logger != nil {
@@ -39,7 +58,7 @@ func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *loggin
 			sf := hpds_node.NewStreamFunction(
 				v.Name,
 				hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", node.Host, node.Port)),
-				hpds_node.WithObserveDataTags(v.DataTag),
+				hpds_node.WithObserveDataTags(frame.Tag(v.DataTag)),
 				hpds_node.WithCredential(node.Token),
 			)
 			err = sf.Connect()
@@ -50,6 +69,12 @@ func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *loggin
 				EndPoint: sf,
 			}
 			must(logger, err)
+			switch v.Name {
+			case "task-log":
+				_ = sf.SetHandler(TaskLogNotificationHandler)
+			default:
+
+			}
 			mqList = append(mqList, nodeInfo)
 		default:
 			ap := hpds_node.NewAccessPoint(
@@ -65,7 +90,7 @@ func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *loggin
 				EndPoint: ap,
 			}
 			must(logger, err)
-			ap.SetDataTag(v.DataTag)
+			ap.SetDataTag(frame.Tag(v.DataTag))
 			mqList = append(mqList, nodeInfo)
 		}
 
@@ -91,3 +116,23 @@ func GenerateAndSendData(stream hpds_node.AccessPoint, data []byte, logger *logg
 	time.Sleep(1000 * time.Millisecond)
 	return nil
 }
+
+func TaskLogNotificationHandler(data []byte) (frame.Tag, []byte) {
+	logging.L().Info("任务日志", zap.String("接收数据", string(data)))
+	cmd := new(InstructionReq)
+	err := json.Unmarshal(data, cmd)
+	if err != nil {
+		return 0x0B, []byte(err.Error())
+	}
+	str, _ := json.Marshal(cmd.Payload)
+	for key := range ChannelsMap {
+		//if strings.Contains(key, userEmail) {
+		//	channel := channelsMap[key]
+		//	channel <- string(msgBytes)
+		//}
+
+		channel := ChannelsMap[key]
+		channel <- string(str)
+	}
+	return 0x0B, nil
+}

mq/instruction.go

@@ -9,3 +9,35 @@ type InstructionReq struct {
 	Command int         `json:"command"`
 	Payload interface{} `json:"payload"`
 }
+
+type ModelResult struct {
+	Code int `json:"code"`
+}
+
+type LightweightResult struct {
+	Code       int    `json:"code"`
+	Crack      bool   `json:"crack"`
+	ImgDiscern string `json:"img_discern"`
+	ImgSrc     string `json:"img_src"`
+	Pothole    bool   `json:"pothole"`
+}
+
+type InsigmaResult struct {
+	Code          int            `json:"code"`
+	NumOfDiseases int            `json:"num_of_diseases"`
+	Diseases      []DiseasesInfo `json:"diseases"`
+	Image         string         `json:"image"`
+}
+
+type DiseasesInfo struct {
+	Id    int           `json:"id"`
+	Type  string        `json:"type"`
+	Level string        `json:"level"`
+	Param DiseasesParam `json:"param"`
+}
+
+type DiseasesParam struct {
+	Length   float64 `json:"length"`
+	Area     float64 `json:"area"`
+	MaxWidth string  `json:"max_width"`
+}

pkg/utils/image.go

@@ -0,0 +1,131 @@
+package utils
+
+import (
+	"bytes"
+	"encoding/base64"
+	"golang.org/x/image/bmp"
+	"golang.org/x/image/tiff"
+	"image"
+	"image/color"
+	"image/jpeg"
+	"image/png"
+)
+
+func BuffToImage(in []byte) image.Image {
+	buff := bytes.NewBuffer(in)
+	m, _, _ := image.Decode(buff)
+	return m
+}
+
+// Clip 图片裁剪
+func Clip(in []byte, wi, hi int, equalProportion bool) (out image.Image, imageType string, err error) {
+	buff := bytes.NewBuffer(in)
+	m, imgType, _ := image.Decode(buff)
+	rgbImg := m.(*image.YCbCr)
+	if equalProportion {
+		w := m.Bounds().Max.X
+		h := m.Bounds().Max.Y
+		if w > 0 && h > 0 && wi > 0 && hi > 0 {
+			wi, hi = fixSize(w, h, wi, hi)
+		}
+	}
+	return rgbImg.SubImage(image.Rect(0, 0, wi, hi)), imgType, nil
+}
+
+func fixSize(img1W, img2H, wi, hi int) (new1W, new2W int) {
+	var ( //为了方便计算，将图片的宽转为 float64
+		imgWidth, imgHeight = float64(img1W), float64(img2H)
+		ratio               float64
+	)
+	if imgWidth >= imgHeight {
+		ratio = imgWidth / float64(wi)
+		return int(imgWidth * ratio), int(imgHeight * ratio)
+	}
+	ratio = imgHeight / float64(hi)
+	return int(imgWidth * ratio), int(imgHeight * ratio)
+}
+
+func Gray(in []byte) (out image.Image, err error) {
+	m := BuffToImage(in)
+	bounds := m.Bounds()
+	dx := bounds.Dx()
+	dy := bounds.Dy()
+	newRgba := image.NewRGBA(bounds)
+	for i := 0; i < dx; i++ {
+		for j := 0; j < dy; j++ {
+			colorRgb := m.At(i, j)
+			_, g, _, a := colorRgb.RGBA()
+			gUint8 := uint8(g >> 8)
+			aUint8 := uint8(a >> 8)
+			newRgba.SetRGBA(i, j, color.RGBA{R: gUint8, G: gUint8, B: gUint8, A: aUint8})
+		}
+	}
+	r := image.Rect(0, 0, dx, dy)
+	return newRgba.SubImage(r), nil
+}
+
+func Rotate90(in []byte) image.Image {
+	m := BuffToImage(in)
+	rotate90 := image.NewRGBA(image.Rect(0, 0, m.Bounds().Dy(), m.Bounds().Dx()))
+	// 矩阵旋转
+	for x := m.Bounds().Min.Y; x < m.Bounds().Max.Y; x++ {
+		for y := m.Bounds().Max.X - 1; y >= m.Bounds().Min.X; y-- {
+			//  设置像素点
+			rotate90.Set(m.Bounds().Max.Y-x, y, m.At(y, x))
+		}
+	}
+	return rotate90
+}
+
+// Rotate180 旋转180度
+func Rotate180(in []byte) image.Image {
+	m := BuffToImage(in)
+	rotate180 := image.NewRGBA(image.Rect(0, 0, m.Bounds().Dx(), m.Bounds().Dy()))
+	// 矩阵旋转
+	for x := m.Bounds().Min.X; x < m.Bounds().Max.X; x++ {
+		for y := m.Bounds().Min.Y; y < m.Bounds().Max.Y; y++ {
+			//  设置像素点
+			rotate180.Set(m.Bounds().Max.X-x, m.Bounds().Max.Y-y, m.At(x, y))
+		}
+	}
+	return rotate180
+}
+
+// Rotate270 旋转270度
+func Rotate270(in []byte) image.Image {
+	m := BuffToImage(in)
+	rotate270 := image.NewRGBA(image.Rect(0, 0, m.Bounds().Dy(), m.Bounds().Dx()))
+	// 矩阵旋转
+	for x := m.Bounds().Min.Y; x < m.Bounds().Max.Y; x++ {
+		for y := m.Bounds().Max.X - 1; y >= m.Bounds().Min.X; y-- {
+			// 设置像素点
+			rotate270.Set(x, m.Bounds().Max.X-y, m.At(y, x))
+		}
+	}
+	return rotate270
+
+}
+
+func ImageToBase64(img image.Image, imgType string) string {
+	buff := ImageToBuff(img, imgType)
+	return base64.StdEncoding.EncodeToString(buff.Bytes())
+}
+
+func ImageToBuff(img image.Image, imgType string) *bytes.Buffer {
+	buff := bytes.NewBuffer(nil)
+	switch imgType {
+	case "bmp":
+		imgType = "bmp"
+		_ = bmp.Encode(buff, img)
+	case "png":
+		imgType = "png"
+		_ = png.Encode(buff, img)
+	case "tiff":
+		imgType = "tiff"
+		_ = tiff.Encode(buff, img, nil)
+	default:
+		imgType = "jpeg"
+		_ = jpeg.Encode(buff, img, nil)
+	}
+	return buff
+}