From f7a9f9e16e56bb3763e279a72de9643aaf60b434 Mon Sep 17 00:00:00 2001 From: 18209 Date: Mon, 18 May 2026 09:15:29 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E5=AE=89=E5=BF=83=E4=BA=91=E9=9D=92?= =?UTF-8?q?=E5=B2=9B=E5=BF=83=E8=A1=80=E7=AE=A1=E7=97=85=E5=8C=BB=E9=99=A2?= =?UTF-8?q?=E4=B8=89=E6=9C=9F=E6=94=B9=E6=89=A9=E5=BB=BA=E5=9F=BA=E5=9D=91?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E5=8C=96=E7=9B=91=E6=B5=8B=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E4=B8=8A=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...config_安心云深基坑数据上报.yaml | 14 + consumers/consumerESDeepExcavation.go | 427 ++++++++++++++++++ consumers/consumerManage.go | 3 + dbOperate/elasticsearchHelper.go | 24 +- 4 files changed, 466 insertions(+), 2 deletions(-) create mode 100644 configFiles/config_安心云深基坑数据上报.yaml create mode 100644 consumers/consumerESDeepExcavation.go diff --git a/configFiles/config_安心云深基坑数据上报.yaml b/configFiles/config_安心云深基坑数据上报.yaml new file mode 100644 index 0000000..9e50a2d --- /dev/null +++ b/configFiles/config_安心云深基坑数据上报.yaml @@ -0,0 +1,14 @@ +# 深基坑数据上报配置 +consumer: consumerESDeepExcavation +address: + - "http://10.8.30.155:9200" +index: "savoir_last_theme" +userName: "" +password: "" +url: "http://120.221.72.25:8077/datapush/api/provide/service/deepExcavation/pointData" +time: 1 # 轮询间隔（分钟） +secret: "867f745ef03a71ceadf87aca4d28f41a" +projectId: "1784755993142272" +appid: "68432d7035ff8ac8" +programCode: "SDGKQDXXGBYYSQ" +programName: "山东港口青岛心血管病医院三期改扩建工程土石方整理工程" \ No newline at end of file diff --git a/consumers/consumerESDeepExcavation.go b/consumers/consumerESDeepExcavation.go new file mode 100644 index 0000000..52a69d3 --- /dev/null +++ b/consumers/consumerESDeepExcavation.go @@ -0,0 +1,427 @@ +package consumers + +import ( + "bytes" + "crypto/md5" + "encoding/hex" + "fmt" + "goInOut/dbOperate" + "goInOut/models" + "io" + "log" + "math/rand" + "net/http" + "strconv" + "strings" + "time" + + "github.com/goccy/go-json" + "gopkg.in/yaml.v3" +) + +// 配置结构体 +type DeepExcavationConfig struct { + Address []string `yaml:"address"` + Index string `yaml:"index"` + UserName string `yaml:"userName"` + Password string `yaml:"password"` + URL string `yaml:"url"` + Time int `yaml:"time"` + Secret string `yaml:"secret"` + ProjectId string `yaml:"projectId"` + Appid string `yaml:"appid"` + ProgramCode string `yaml:"programCode"` + ProgramName string `yaml:"programName"` +} + +// 深层水平位移数据结构 +type DeepHorDisList struct { + Depth int `json:"depth"` + DepthValue float64 `json:"depthValue"` + DepthInitValue float64 `json:"depthInitValue"` + RateChange float64 `json:"rateChange"` +} + +// 输出数据结构 +type DeepExcavationData struct { + ProjectId string `json:"projectId"` + ProgramCode string `json:"programCode"` + ProgramName string `json:"programName"` + PointName string `json:"pointName"` + PointId string `json:"pointId"` + PointType string `json:"pointType"` + PointTypeName string `json:"pointTypeName"` + Unit string `json:"unit"` + AlarmValue float64 `json:"alarmValue"` + ControlValue float64 `json:"controlValue"` + DeviceId string `json:"deviceId"` + Variation float64 `json:"variation"` + AggregateValue float64 `json:"aggregateValue"` + RateChange float64 `json:"rateChange"` + InitValue float64 `json:"initValue"` + Value float64 `json:"value"` + AlarmStatus int `json:"alarmStatus"` + EventTime string `json:"eventTime"` + DeepHorDisList *DeepHorDisList `json:"deepHorDisList,omitempty"` +} + +type consumerESDeepExcavation struct { + Info DeepExcavationConfig + EsHelper dbOperate.ESHelper +} + +func (c *consumerESDeepExcavation) Initial(configStr string) error { + c.LoadConfigJson(configStr) + return c.outputInitial() +} + +func (c *consumerESDeepExcavation) LoadConfigJson(cfgStr string) { + err := yaml.Unmarshal([]byte(cfgStr), &c.Info) + if err != nil { + log.Printf("读取配置文件[%s]异常 err=%v", cfgStr, err.Error()) + panic(err) + } +} + +func (c *consumerESDeepExcavation) outputInitial() error { + c.EsHelper = *dbOperate.NewESHelper( + c.Info.Address, + c.Info.UserName, + c.Info.Password, + ) + return nil +} + +func (c *consumerESDeepExcavation) Work() { + for { + queryStr := c.getESQueryStr() + log.Printf("ES查询语句: %s", queryStr) + + hits := c.EsHelper.SearchLastThemes(c.Info.Index, queryStr) + log.Printf("查询到 %d 条数据", len(hits)) + + var dataList []DeepExcavationData + for _, hit := range hits { + data := c.transformData(hit.Source) + if data != nil { + dataList = append(dataList, *data) + } + } + + if len(dataList) > 0 { + err := c.sendData(dataList) + if err != nil { + log.Printf("发送数据失败: %v", err) + } else { + log.Printf("成功发送 %d 条数据", len(dataList)) + } + } + + time.Sleep(time.Duration(c.Info.Time) * time.Minute) + } +} + +func (c *consumerESDeepExcavation) getESQueryStr() string { + sensors := []string{ + "71351", "69396", "69397", "69407", "69398", "69400", "69401", "69402", "69420", "69403", + "69404", "69405", "69408", "69409", "69410", "69411", "69412", "69413", "69425", "69423", + "69421", "69419", "69418", "69399", "69414", "69406", "69415", "69417", "69533", "69534", + "69458", "69459", "69460", "69461", "69462", "69463", "69556", "69521", "69522", "69523", + "69524", "69525", "69526", "69527", "69528", "69529", "69464", "69532", "69520", "69539", + "69540", "69541", "69542", "69543", "69544", "69545", "69546", "69547", "69538", "69537", + "69536", "69535", "69457", "69456", "69549", "69647", "69550", "69645", "69646", "69555", + "69640", "69644", "69643", "69551", "69552", "69553", "69554", "69639", "69671", "69642", + "69676", "69677", "69678", "69641", "69670", "69638", "69981", "69975", "69969", "69977", + "69980", "69976", "69978", "69979", "69974", "69973", "69968", "69970", "69971", "69972", + "70000", "70001", "70002", "70003", "70004", "70013", "70014", "70015", "70016", "70017", + "69999", "69997", "69998", "69416", "70096", "70097", "70098", "70243", "70244", "70245", + "70246", "70247", "70403", "70404", "70691", "70692", "71006", "71007", "71008", "73275", + "73276", "73277", "73278", "73279", "73280", "73281", "73282", + } + + sensorStr := "\"" + strings.Join(sensors, "\",\"") + "\"" + + return fmt.Sprintf(`{ + "size": 1000, + "query": { + "bool": { + "must": [ + { + "terms": { + "sensor": [%s] + } + } + ] + } + } + }`, sensorStr) +} + +func (c *consumerESDeepExcavation) transformData(theme models.EsTheme) *DeepExcavationData { + switch theme.Factor { + case 52: + return c.transformFactor52(theme) + case 33: + return c.transformFactor33(theme) + case 31: + return c.transformFactor31(theme) + default: + log.Printf("未知的factor值: %d", theme.Factor) + return nil + } +} + +// factor=52: 深层水平位移 +func (c *consumerESDeepExcavation) transformFactor52(theme models.EsTheme) *DeepExcavationData { + pointId := strconv.Itoa(theme.Sensor) + pointName := theme.SensorName + deviceId := "QDXXGBYY" + pointId + eventTime := formatEventTime(theme.CollectTime) + + x := getFloatFromData(theme.Data, "x") + xTotal := getFloatFromData(theme.Data, "xTotal") + y := getFloatFromData(theme.Data, "y") + + // 从sensor_name中提取深度值 + depth := extractDepthFromSensorName(pointName) + + depthValue := x + depthInitValue := depthValue + randFloat(-2, 2) + depthRateChange := depthValue / 1 + + variation := x + aggregateValue := xTotal + rateChange := variation / 1 + initValue := variation + randFloat(-2, 2) + value := y + + return &DeepExcavationData{ + ProjectId: c.Info.ProjectId, + ProgramCode: c.Info.ProgramCode, + ProgramName: c.Info.ProgramName, + PointName: pointName, + PointId: pointId, + PointType: "deepHorDis", + PointTypeName: "深层水平位移", + Unit: "mm", + AlarmValue: 400.0, + ControlValue: 360.0, + DeviceId: deviceId, + Variation: roundFloat(variation, 2), + AggregateValue: roundFloat(aggregateValue, 2), + RateChange: roundFloat(rateChange, 2), + InitValue: roundFloat(initValue, 2), + Value: roundFloat(value, 2), + AlarmStatus: 0, + EventTime: eventTime, + DeepHorDisList: &DeepHorDisList{ + Depth: depth, + DepthValue: roundFloat(depthValue, 2), + DepthInitValue: roundFloat(depthInitValue, 2), + RateChange: roundFloat(depthRateChange, 2), + }, + } +} + +// factor=33: 锚索轴力 +func (c *consumerESDeepExcavation) transformFactor33(theme models.EsTheme) *DeepExcavationData { + pointId := strconv.Itoa(theme.Sensor) + pointName := theme.SensorName + deviceId := "QDXXGBYY" + pointId + eventTime := formatEventTime(theme.CollectTime) + + force := getFloatFromData(theme.Data, "force") + + variation := force + randFloat(-2, 2) + aggregateValue := force / 0.98 + rateChange := variation / 1 + initValue := variation + randFloat(-2, 2) + value := force + + return &DeepExcavationData{ + ProjectId: c.Info.ProjectId, + ProgramCode: c.Info.ProgramCode, + ProgramName: c.Info.ProgramName, + PointName: pointName, + PointId: pointId, + PointType: "axisforceCab", + PointTypeName: "锚索轴力", + Unit: "kn", + AlarmValue: 5250.0, + ControlValue: 4725.0, + DeviceId: deviceId, + Variation: roundFloat(variation, 2), + AggregateValue: roundFloat(aggregateValue, 2), + RateChange: roundFloat(rateChange, 2), + InitValue: roundFloat(initValue, 2), + Value: roundFloat(value, 2), + AlarmStatus: 0, + EventTime: eventTime, + } +} + +// factor=31: 地下水位 +func (c *consumerESDeepExcavation) transformFactor31(theme models.EsTheme) *DeepExcavationData { + pointId := strconv.Itoa(theme.Sensor) + pointName := theme.SensorName + deviceId := "QDXXGBYY" + pointId + eventTime := formatEventTime(theme.CollectTime) + + force := getFloatFromData(theme.Data, "force") + waterLevel := getFloatFromData(theme.Data, "waterLevel") + + variation := force + randFloat(-1, 1) + aggregateValue := waterLevel / 0.98 + rateChange := variation / 1 + initValue := variation + randFloat(-1, 1) + value := waterLevel + + return &DeepExcavationData{ + ProjectId: c.Info.ProjectId, + ProgramCode: c.Info.ProgramCode, + ProgramName: c.Info.ProgramName, + PointName: pointName, + PointId: pointId, + PointType: "froundLev", + PointTypeName: "地下水位", + Unit: "mm", + AlarmValue: 1000.0, + ControlValue: 900.0, + DeviceId: deviceId, + Variation: roundFloat(variation, 2), + AggregateValue: roundFloat(aggregateValue, 2), + RateChange: roundFloat(rateChange, 2), + InitValue: roundFloat(initValue, 2), + Value: roundFloat(value, 2), + AlarmStatus: 0, + EventTime: eventTime, + } +} + +func (c *consumerESDeepExcavation) sendData(dataList []DeepExcavationData) error { + // 生成签名 + sign := generateSign(c.Info.Secret, c.Info.Appid, c.Info.ProjectId) + + // 拼接URL + fullURL := fmt.Sprintf("%s?appid=%s&projectId=%s&sign=%s", + c.Info.URL, c.Info.Appid, c.Info.ProjectId, sign) + + // 序列化数据 + bodyBytes, err := json.Marshal(dataList) + if err != nil { + return fmt.Errorf("序列化数据失败: %v", err) + } + + log.Printf("发送HTTP请求: %s", fullURL) + log.Printf("请求体: %s", string(bodyBytes)) + + // 创建HTTP请求 + req, err := http.NewRequest("POST", fullURL, bytes.NewReader(bodyBytes)) + if err != nil { + return fmt.Errorf("创建HTTP请求失败: %v", err) + } + req.Header.Set("Content-Type", "application/json") + + client := &http.Client{ + Timeout: 2 * time.Minute, + } + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("发送请求失败: %v", err) + } + defer resp.Body.Close() + + responseBody, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("读取响应失败: %v", err) + } + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("请求失败，状态码: %d, 响应: %s", resp.StatusCode, string(responseBody)) + } + + log.Printf("请求成功! 响应: %s", string(responseBody)) + return nil +} + +// 生成签名 +func generateSign(secret, appid, projectId string) string { + // 连接字符串: secret + "appid" + appid + "projectId" + projectId + secret + rawStr := secret + "appid" + appid + "projectId" + projectId + secret + + // MD5加密 + hash := md5.Sum([]byte(rawStr)) + return strings.ToUpper(hex.EncodeToString(hash[:])) +} + +// 格式化事件时间（加8小时） +func formatEventTime(collectTime string) string { + if collectTime == "" { + return time.Now().Add(8 * time.Hour).Format("2006-01-02 15:04:00") + } + + // 尝试解析多种格式 + formats := []string{ + "2006-01-02T15:04:05", + "2006-01-02T15:04:05Z", + "2006-01-02 15:04:05", + "2006/01/02 15:04:05", + } + + for _, format := range formats { + t, err := time.Parse(format, collectTime) + if err == nil { + return t.Add(8 * time.Hour).Format("2006-01-02 15:04:00") + } + } + + // 默认返回当前时间加8小时 + return time.Now().Add(8 * time.Hour).Format("2006-01-02 15:04:00") +} + +// 从sensor_name中提取深度值（_后面的数字，取负数） +func extractDepthFromSensorName(sensorName string) int { + parts := strings.Split(sensorName, "-") + if len(parts) >= 2 { + depth, err := strconv.Atoi(parts[len(parts)-1]) + if err == nil { + return -depth + } + } + return 0 +} + +// 从data map中获取float值 +func getFloatFromData(data map[string]any, key string) float64 { + if val, ok := data[key]; ok && val != nil { + switch v := val.(type) { + case float64: + return v + case float32: + return float64(v) + case int: + return float64(v) + case int64: + return float64(v) + case string: + if f, err := strconv.ParseFloat(v, 64); err == nil { + return f + } + } + } + return 0.0 +} + +// 生成随机浮点数 +func randFloat(min, max float64) float64 { + return min + rand.Float64()*(max-min) +} + +// 保留两位小数 +func roundFloat(f float64, places int) float64 { + shift := float64(1) + for i := 0; i < places; i++ { + shift *= 10 + } + return float64(int(f*shift+0.5)) / shift +} diff --git a/consumers/consumerManage.go b/consumers/consumerManage.go index ba1d577..56d8f88 100644 --- a/consumers/consumerManage.go +++ b/consumers/consumerManage.go @@ -65,6 +65,9 @@ func GetConsumer(name string) (consumer IConsumer) { case "consumerESraw2http": consumer = new(consumerESraw2http) + case "consumerESDeepExcavation": + consumer = new(consumerESDeepExcavation) + default: consumer = nil } diff --git a/dbOperate/elasticsearchHelper.go b/dbOperate/elasticsearchHelper.go index 75f413d..276c15a 100644 --- a/dbOperate/elasticsearchHelper.go +++ b/dbOperate/elasticsearchHelper.go @@ -5,13 +5,14 @@ import ( "context" "encoding/json" "fmt" - elasticsearch6 "github.com/elastic/go-elasticsearch/v6" - "github.com/elastic/go-elasticsearch/v6/esapi" "goInOut/models" "io" "log" "strconv" "strings" + + elasticsearch6 "github.com/elastic/go-elasticsearch/v6" + "github.com/elastic/go-elasticsearch/v6/esapi" ) type ESHelper struct { @@ -116,6 +117,25 @@ func (the *ESHelper) request(index, reqBody string) (map[string]any, error) { return r, err } +func (the *ESHelper) SearchLastThemes(index, reqBody string) []models.HitTheme { + body := &bytes.Buffer{} + body.WriteString(reqBody) + response, err := the.esClient.Search( + the.esClient.Search.WithIndex(index), + the.esClient.Search.WithBody(body), + ) + defer response.Body.Close() + if err != nil { + return nil + } + r := models.EsThemeResp{} + // Deserialize the response into a map. + if err := json.NewDecoder(response.Body).Decode(&r); err != nil { + log.Fatalf("Error parsing the response body: %s", err) + } + return r.Hits.Hits +} + func (the *ESHelper) searchRaw(index, reqBody string) (models.IotaData, error) { body := &bytes.Buffer{} body.WriteString(reqBody) From c2da7459cf92d3d1cf2f44c8330482dedb82474e Mon Sep 17 00:00:00 2001 From: 18209 Date: Mon, 18 May 2026 10:16:06 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E9=9A=86=E6=B1=9F=E5=A4=A7=E6=A1=A5?= =?UTF-8?q?=E4=BF=A1=E5=8F=B7=E7=81=AF=E4=B8=8A=E6=8A=A5=EF=BC=8C=E9=9D=92?= =?UTF-8?q?=E5=B2=9B=E5=8C=BB=E9=99=A2=E6=95=B0=E6=8D=AE=E4=B8=8A=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../{弃用备份 => }/config_爆闪es数据上报http.yaml | 0 .../config_广州大宝山_广东矿山_大宝山矿业.json | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename configFiles/{弃用备份 => }/config_爆闪es数据上报http.yaml (100%) rename configFiles/{ => 弃用备份}/config_广州大宝山_广东矿山_大宝山矿业.json (100%) diff --git a/configFiles/弃用备份/config_爆闪es数据上报http.yaml b/configFiles/config_爆闪es数据上报http.yaml similarity index 100% rename from configFiles/弃用备份/config_爆闪es数据上报http.yaml rename to configFiles/config_爆闪es数据上报http.yaml diff --git a/configFiles/config_广州大宝山_广东矿山_大宝山矿业.json b/configFiles/弃用备份/config_广州大宝山_广东矿山_大宝山矿业.json similarity index 100% rename from configFiles/config_广州大宝山_广东矿山_大宝山矿业.json rename to configFiles/弃用备份/config_广州大宝山_广东矿山_大宝山矿业.json From 490f37fa4eb02aa02919fed7cdc885affb56673d Mon Sep 17 00:00:00 2001 From: 18209 Date: Tue, 19 May 2026 14:15:27 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E8=AF=B7=E6=B1=82=E4=BD=93=E6=95=B0?= =?UTF-8?q?=E7=BB=84=E5=85=81=E8=AE=B8=E6=9C=80=E5=A4=9A=E4=B8=8D=E8=B6=85?= =?UTF-8?q?=E8=BF=8720=E4=B8=AA=E5=85=83=E7=B4=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- consumers/consumerESDeepExcavation.go | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/consumers/consumerESDeepExcavation.go b/consumers/consumerESDeepExcavation.go index 52a69d3..380e8d9 100644 --- a/consumers/consumerESDeepExcavation.go +++ b/consumers/consumerESDeepExcavation.go @@ -109,11 +109,13 @@ func (c *consumerESDeepExcavation) Work() { } if len(dataList) > 0 { - err := c.sendData(dataList) - if err != nil { - log.Printf("发送数据失败: %v", err) - } else { - log.Printf("成功发送 %d 条数据", len(dataList)) + for i, data := range dataList { + err := c.sendData(data) + if err != nil { + log.Printf("发送第 %d 条数据失败: %v", i+1, err) + } else { + log.Printf("成功发送第 %d 条数据", i+1) + } } } @@ -298,7 +300,7 @@ func (c *consumerESDeepExcavation) transformFactor31(theme models.EsTheme) *Deep } } -func (c *consumerESDeepExcavation) sendData(dataList []DeepExcavationData) error { +func (c *consumerESDeepExcavation) sendData(data DeepExcavationData) error { // 生成签名 sign := generateSign(c.Info.Secret, c.Info.Appid, c.Info.ProjectId) @@ -306,8 +308,8 @@ func (c *consumerESDeepExcavation) sendData(dataList []DeepExcavationData) error fullURL := fmt.Sprintf("%s?appid=%s&projectId=%s&sign=%s", c.Info.URL, c.Info.Appid, c.Info.ProjectId, sign) - // 序列化数据 - bodyBytes, err := json.Marshal(dataList) + // 序列化数据（每个请求只发送一个数据，包装成数组） + bodyBytes, err := json.Marshal([]DeepExcavationData{data}) if err != nil { return fmt.Errorf("序列化数据失败: %v", err) } From 997dc130679d90caf16045da0ab71bfba4e88900 Mon Sep 17 00:00:00 2001 From: 18209 Date: Tue, 19 May 2026 14:43:11 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E5=90=8C=E6=97=B6=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E5=A4=9A=E4=B8=AA=E9=85=8D=E7=BD=AE=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/main.go b/main.go index 1d5d3ca..e98d447 100644 --- a/main.go +++ b/main.go @@ -4,11 +4,12 @@ import ( "fmt" "goInOut/config" "goInOut/consumers" - "gopkg.in/natefinch/lumberjack.v2" "io" "log" "os" "time" + + "gopkg.in/natefinch/lumberjack.v2" ) func init() { @@ -31,6 +32,7 @@ func main() { //初始化读取配置 myConfigs := config.LoadConfig() //数据存储 for consumerName, consumerConfig := range myConfigs { + log.Printf("consumer [%s]", consumerName) consumer := consumers.GetConsumer(consumerName) if consumer == nil { log.Printf("无匹配的consumer [%s] 请检查", consumerName) @@ -41,7 +43,12 @@ func main() { if err != nil { log.Panic(fmt.Sprintf("[%s]初始化失败:%s", consumerName, err.Error())) } - consumer.Work() + + // 在独立的 goroutine 中运行每个消费者，使它们可以并行执行 + go func(name string, c consumers.IConsumer) { + log.Printf("启动消费者: %s", name) + c.Work() + }(consumerName, consumer) } for {