no message

common_utils/log/logger.go

@@ -2,6 +2,7 @@ package log
 
 import (
 	"fmt"
+	"github.com/natefinch/lumberjack"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"os"
@@ -12,14 +13,14 @@ var logger *zap.Logger
 
 func InitLogger(logpath string, loglevel string) {
 	// 日志分割
-	//hook := lumberjack.Logger{
-	//	Filename:   logpath, // 日志文件路径，默认 os.TempDir()
-	//	MaxSize:    200,     // 每个日志文件保存10M，默认 100M
-	//	MaxBackups: 20,      // 保留30个备份，默认不限
-	//	MaxAge:     15,      // 保留7天，默认不限
-	//	Compress:   false,   // 是否压缩，默认不压缩
-	//}
-	//write := zapcore.AddSync(&hook)
+	hook := lumberjack.Logger{
+		Filename:   logpath, // 日志文件路径，默认 os.TempDir()
+		MaxSize:    200,     // 每个日志文件保存10M，默认 100M
+		MaxBackups: 20,      // 保留30个备份，默认不限
+		MaxAge:     15,      // 保留7天，默认不限
+		Compress:   false,   // 是否压缩，默认不压缩
+	}
+	write := zapcore.AddSync(&hook)
 	// 设置日志级别
 	// debug 可以打印出 info debug warn
 	// info  级别可以打印 warn info
@@ -56,8 +57,8 @@ func InitLogger(logpath string, loglevel string) {
 	core := zapcore.NewCore(
 		// zapcore.NewConsoleEncoder(encoderConfig),
 		zapcore.NewJSONEncoder(encoderConfig),
-		zapcore.NewMultiWriteSyncer(zapcore.AddSync(os.Stdout)),
-		//zapcore.NewMultiWriteSyncer(zapcore.AddSync(os.Stdout), zapcore.AddSync(write)), // 打印到控制台和文件
+		//zapcore.NewMultiWriteSyncer(zapcore.AddSync(os.Stdout)),
+		zapcore.NewMultiWriteSyncer(zapcore.AddSync(os.Stdout), zapcore.AddSync(write)), // 打印到控制台和文件
 		//write, //文件
 		level,
 	)
@@ -66,15 +67,16 @@ func InitLogger(logpath string, loglevel string) {
 	// 开启文件及行号
 	development := zap.Development()
 	// 设置初始化字段,如：添加一个服务器名称
-	//filed := zap.Fields(zap.String("serviceName", "serviceName"))
+	filed := zap.Fields(zap.String("serviceName", "serviceName"))
 	// 构造日志
-	logger = zap.New(core, caller, development)
+	logger = zap.New(core, caller, filed, development)
 	logger.Info("DefaultLogger init success")
 }
 
 func formatEncodeTime(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
 	enc.AppendString(fmt.Sprintf("%d-%02d-%02d %02d:%02d:%02d", t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second()))
 }
+
 func Field(key string, val interface{}) zap.Field {
 	return zap.Any(key, val)
 }

jy_publishing/config.json

@@ -26,7 +26,7 @@
   },
   "nsq_jy": {
     "addr": "192.168.3.207:4150",
-    "topic": "bidding-save",
+    "topic": "jyinfo",
     "channel": "data-dispose",
     "concurrent": 1
   },

jy_publishing/go.mod

@@ -6,11 +6,13 @@ require (
 	app.yhyue.com/BP/servicerd v0.0.0-20201203055056-87643512f867
 	github.com/Chain-Zhang/pinyin v0.1.3
 	github.com/golang/protobuf v1.5.2
+	github.com/mitchellh/mapstructure v1.1.2
 	github.com/nsqio/go-nsq v1.1.0
 	github.com/zeromicro/go-zero v1.3.2
 	google.golang.org/grpc v1.44.0
 	google.golang.org/protobuf v1.27.1
 	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
+	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	utils v0.0.0
 )
 

jy_publishing/go.sum

@@ -42,6 +42,7 @@ github.com/Azure/go-autorest/autorest/mocks v0.4.0/go.mod h1:LTp+uSrOhSkaKrUy935
 github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
 github.com/Azure/go-autorest/logger v0.2.0/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8=
 github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/Chain-Zhang/pinyin v0.1.3 h1:RzErNyNwVa8z2sOLCuXSOtVdY/AsARb8mBzI2p2qtnE=
@@ -342,6 +343,7 @@ github.com/mattn/go-sqlite3 v1.14.2/go.mod h1:JIl7NbARA7phWnGvh0LKTyg7S9BA+6gx71
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
@@ -355,6 +357,7 @@ github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8m
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw=
+github.com/natefinch/lumberjack v2.0.0+incompatible h1:4QJd3OLAMgj7ph+yZTuX13Ld4UpgHp07nNdFX7mqFfM=
 github.com/natefinch/lumberjack v2.0.0+incompatible/go.mod h1:Wi9p2TTF5DG5oU+6YfsmYQpsTIOm0B1VNzQg9Mw6nPk=
 github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32/go.mod h1:9wM+0iRr9ahx58uYLpLIr5fm8diHn0JbqRycJi6w0Ms=
 github.com/nsqio/go-nsq v1.1.0 h1:PQg+xxiUjA7V+TLdXw7nVrJ5Jbl3sN86EhGCQj4+FYE=
@@ -844,6 +847,8 @@ gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 h1:VpOs+IwYnYBaFnrNAeB8UUWtL3vEUnzSCL1nVjPhqrw=
 gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=
+gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8=
+gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
 gopkg.in/olivere/elastic.v2 v2.0.61 h1:7cpl3MW8ysa4GYFBXklpo5mspe4NK0rpZTdyZ+QcD4U=
 gopkg.in/olivere/elastic.v2 v2.0.61/go.mod h1:CTVyl1gckiFw1aLZYxC00g3f9jnHmhoOKcWF7W3c6n4=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=

jy_publishing/main.go

@@ -2,11 +2,11 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"github.com/zeromicro/go-zero/core/discov"
 	"github.com/zeromicro/go-zero/zrpc"
 	ms "jy_publishing/megaloscope"
 	nsq "jy_publishing/nsq"
-	pb "jy_publishing/proto"
 	"net"
 	"strings"
 	"utils"
@@ -74,8 +74,8 @@ func init() {
 		},
 	})
 
-	initUdp()
-	initNsq()
+	//initUdp()
+	//initNsq()
 }
 
 func initNsq() {
@@ -91,7 +91,7 @@ func initNsq() {
 	cofAtts := Sysconfig["nsq_attachment"].(map[string]interface{})
 	MProducer, _ = nsq.NewProducer(util.ObjToString(cofAtts["addr"]), util.ObjToString(cofAtts["topic"]), true)
 	MCAtts, _ = nsq.NewConsumer(&nsq.Cconfig{
-		IsJsonEncode: true, //与生产者配置对应，设为true会取第1个字节进行类型判断
+		IsJsonEncode: false, //与生产者配置对应，设为true会取第1个字节进行类型判断
 		Addr:         util.ObjToString(cofAtts["addr"]),
 		ConnectType:  0, //默认连接nsqd
 		Topic:        util.ObjToString(cofAtts["topic-result"]),
@@ -108,7 +108,7 @@ func initUdp() {
 }
 
 func main() {
-	go jyNsqMethod()
+	//go jyNsqMethod()
 	go attsNsqMethod()
 
 	attsMap := map[string]interface{}{"1": map[string]interface{}{
@@ -117,34 +117,38 @@ func main() {
 		"ftype":    "docx",
 	}}
 	other := map[string]interface{}{
-		"id":      "112",
+		"id":      "113",
 		"action":  "1",
 		"msgType": "1",
-		"title":   []string{},
+		"title":   []string{"汉"},
 		"detail":  []string{},
 	}
-	otherJson, _ := json.Marshal(other)
-	var attsArr []*pb.Request
+	//otherJson, _ := json.Marshal(other)
+	//var attsArr []*pb.Request
+	var attsArr []map[string]interface{}
 	for _, m := range attsMap {
 		m1 := m.(map[string]interface{})
-		attsArr = append(attsArr, &pb.Request{
-			FileUrl:     util.ObjToString(m1["fid"]),
-			FileName:    util.ObjToString(m1["filename"]),
-			FileType:    util.ObjToString(m1["ftype"]),
-			ReturnType:  1,
-			ExtractType: 0,
+		//attsArr = append(attsArr, &pb.Request{
+		//	FileUrl:     util.ObjToString(m1["fid"]),
+		//	FileName:    util.ObjToString(m1["filename"]),
+		//	FileType:    util.ObjToString(m1["ftype"]),
+		//	ReturnType:  1,
+		//	ExtractType: 1,
+		//})
+		attsArr = append(attsArr, map[string]interface{}{
+			"fileUrl":     util.ObjToString(m1["fid"]),
+			"fileName":    util.ObjToString(m1["filename"]),
+			"fileType":    util.ObjToString(m1["ftype"]),
+			"returnType":  1,
+			"extractType": 0,
 		})
 	}
-	msginfo := &pb.FileRequest{
-		Message: attsArr,
-		Other:   string(otherJson),
-		Topic:   "data-attachment",
+	msginfo := map[string]interface{}{
+		"message": attsArr,
+		"other":   other,
+		"topic":   "attachment-txt",
 	}
-	//msginfo := map[string]interface{}{
-	//	"meassage": attsArr,
-	//	"other":    string(otherJson),
-	//	"topic":    "data-attachment",
-	//}
+	util.Debug(msginfo)
 	_ = MProducer.Publish(msginfo)
 
 	c := make(chan bool, 1)
@@ -157,6 +161,7 @@ func jyNsqMethod() {
 	for {
 		select {
 		case obj := <-MCJy.Ch: //从通道读取即可
+			fmt.Println("---", obj)
 			taskInfo(obj)
 		}
 	}
@@ -168,8 +173,10 @@ func attsNsqMethod() {
 	for {
 		select {
 		case obj := <-MCAtts.Ch:
-			util.Debug(obj)
-			//taskAtts(obj)
+			var resp *AttsResponse
+			_ = json.Unmarshal(obj.([]byte), &resp)
+			fmt.Println("---", resp)
+			taskAtts(resp)
 		}
 	}
 }
@@ -177,17 +184,24 @@ func attsNsqMethod() {
 func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 	switch act {
 	case util.OP_TYPE_DATA: //测试接收
-		var rep map[string]interface{}
-		err := json.Unmarshal(data, &rep)
-		log.Info("udp receive...", log.Field("data", rep))
+		var resp map[string]interface{}
+		err := json.Unmarshal(data, &resp)
+		log.Info("udp receive...", log.Field("data", resp))
 		if err != nil {
 			//go Udpclient.WriteUdp([]byte{}, mu.OP_NOOP, ra) //回应上一个节点
-		} else {
-			//by, _ := json.Marshal(map[string]interface{}{
-			//	"taskid": rep["taskid"],
-			//	"stype":  rep["stype"],
-			//})
-			//go Udpclient.WriteUdp(by, mu.OP_NOOP, ra) //回应上一个节点
+		} else if resp != nil {
+			key, _ := resp["key"].(string)
+			if key == "" {
+				key = "udpok"
+			}
+			go UdpClient.WriteUdp([]byte(key), util.OP_NOOP, ra)
+			tasktype, _ := resp["stype"].(string)
+			switch tasktype {
+			case "jyfb_data_over":
+				go func() {
+					DelMethod(resp["infoid"].(string))
+				}()
+			}
 		}
 	case util.OP_NOOP: //下个节点回应
 		log.Info("接收回应：", log.Field("data", string(data)))

jy_publishing/nsq/consumer.go

@@ -2,6 +2,7 @@ package gonsq
 
 import (
 	"encoding/json"
+	"log"
 	"strings"
 	"time"
 
@@ -30,6 +31,7 @@ func (c *Consumer) HandleMessage(msg *nsq.Message) error {
 	if c.IsJsonEncode {
 		if len(msg.Body) > 1 {
 			var err error
+			log.Println("1--" + string(msg.Body))
 			switch msg.Body[0] {
 			case 0x00:
 				var obj interface{}
@@ -43,6 +45,12 @@ func (c *Consumer) HandleMessage(msg *nsq.Message) error {
 				if err == nil && obj != nil {
 					c.Ch <- obj
 				}
+			default:
+				var obj interface{}
+				err = json.Unmarshal(msg.Body, &obj)
+				if err == nil && obj != nil {
+					c.Ch <- obj
+				}
 			}
 			return err
 		}

jy_publishing/proto/fileText.pb.go

@@ -37,8 +37,8 @@ type FileRequest struct {
 	unknownFields protoimpl.UnknownFields
 
 	Message []*Request `protobuf:"bytes,1,rep,name=message,proto3" json:"message,omitempty"` //文件名称
-	Other   string     `protobuf:"bytes,2,opt,name=Other,proto3" json:"Other,omitempty"`
-	Topic   string     `protobuf:"bytes,3,opt,name=Topic,proto3" json:"Topic,omitempty"`
+	Other   string     `protobuf:"bytes,2,opt,name=other,proto3" json:"other,omitempty"`     //信息id
+	Topic   string     `protobuf:"bytes,3,opt,name=topic,proto3" json:"topic,omitempty"`
 }
 
 func (x *FileRequest) Reset() {
@@ -104,7 +104,7 @@ type Request struct {
 	FileBytes   []byte `protobuf:"bytes,3,opt,name=fileBytes,proto3" json:"fileBytes,omitempty"`      //文件字节流
 	FileType    string `protobuf:"bytes,4,opt,name=fileType,proto3" json:"fileType,omitempty"`        //文件类型 pdf ,xls ,doc
 	ReturnType  int32  `protobuf:"varint,5,opt,name=returnType,proto3" json:"returnType,omitempty"`   //文件类型 {0:url,1:content,2:url+content} 默认0
-	ExtractType int32  `protobuf:"varint,6,opt,name=extractType,proto3" json:"extractType,omitempty"` //解析文件类型{0:url,1:fileBytes,2:oss} 默认2
+	ExtractType int32  `protobuf:"varint,6,opt,name=extractType,proto3" json:"extractType,omitempty"` //解析文件类型{2:oss，1:fileBytes,2:url} 默认2
 }
 
 func (x *Request) Reset() {
@@ -187,7 +187,7 @@ type FileResponse struct {
 	unknownFields protoimpl.UnknownFields
 
 	Result []*Result `protobuf:"bytes,1,rep,name=result,proto3" json:"result,omitempty"` //返回结果
-	Other  string    `protobuf:"bytes,2,opt,name=Other,proto3" json:"Other,omitempty"`
+	Other  string    `protobuf:"bytes,2,opt,name=other,proto3" json:"other,omitempty"`   //信息id
 }
 
 func (x *FileResponse) Reset() {
@@ -245,7 +245,7 @@ type Result struct {
 	TextContent string `protobuf:"bytes,2,opt,name=textContent,proto3" json:"textContent,omitempty"` //文本内容
 	TextUrl     string `protobuf:"bytes,3,opt,name=textUrl,proto3" json:"textUrl,omitempty"`         //文本URL
 	FilePath    string `protobuf:"bytes,4,opt,name=filePath,proto3" json:"filePath,omitempty"`       //文件路径
-	ErrorState  string `protobuf:"bytes,5,opt,name=errorState,proto3" json:"errorState,omitempty"`   //状态{200:解析成功，300:下载失败，400:解析失败}
+	ErrorState  string `protobuf:"bytes,5,opt,name=errorState,proto3" json:"errorState,omitempty"`   //错误状态{0:文件下载失败，1：文件解析失败}
 }
 
 func (x *Result) Reset() {
@@ -323,9 +323,9 @@ var file_fileText_proto_rawDesc = []byte{
 	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x28, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
 	0x65, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e,
 	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
-	0x12, 0x14, 0x0a, 0x05, 0x4f, 0x74, 0x68, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x05, 0x4f, 0x74, 0x68, 0x65, 0x72, 0x12, 0x14, 0x0a, 0x05, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x18,
-	0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x22, 0xbb, 0x01, 0x0a,
+	0x12, 0x14, 0x0a, 0x05, 0x6f, 0x74, 0x68, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x05, 0x6f, 0x74, 0x68, 0x65, 0x72, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x22, 0xbb, 0x01, 0x0a,
 	0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x69, 0x6c, 0x65,
 	0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x66, 0x69, 0x6c, 0x65,
 	0x4e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x66, 0x69, 0x6c, 0x65, 0x55, 0x72, 0x6c, 0x18,
@@ -341,8 +341,8 @@ var file_fileText_proto_rawDesc = []byte{
 	0x6c, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x06, 0x72, 0x65,
 	0x73, 0x75, 0x6c, 0x74, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x70, 0x72, 0x6f,
 	0x74, 0x6f, 0x2e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x06, 0x72, 0x65, 0x73, 0x75, 0x6c,
-	0x74, 0x12, 0x14, 0x0a, 0x05, 0x4f, 0x74, 0x68, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x05, 0x4f, 0x74, 0x68, 0x65, 0x72, 0x22, 0x9c, 0x01, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75,
+	0x74, 0x12, 0x14, 0x0a, 0x05, 0x6f, 0x74, 0x68, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x05, 0x6f, 0x74, 0x68, 0x65, 0x72, 0x22, 0x9c, 0x01, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75,
 	0x6c, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x69, 0x6c, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x66, 0x69, 0x6c, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x20,
 	0x0a, 0x0b, 0x74, 0x65, 0x78, 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x02, 0x20,

jy_publishing/proto/fileText.proto

@@ -4,8 +4,8 @@ package proto;
 
 message FileRequest {
   repeated Request message = 1; //文件名称
-  string Other = 2;
-  string Topic = 3;
+  string other = 2;     //信息id
+  string topic = 3;
 }
 
 message Request {
@@ -14,12 +14,12 @@ message Request {
   bytes fileBytes = 3;     //文件字节流
   string fileType = 4;    //文件类型 pdf ,xls ,doc
   int32 returnType = 5;    //文件类型 {0:url,1:content,2:url+content} 默认0
-  int32 extractType = 6;  //解析文件类型{0:url,1:fileBytes,2:oss} 默认2
+  int32 extractType = 6;  //解析文件类型{2:oss，1:fileBytes,2:url} 默认2
 }
 
 message FileResponse {
   repeated Result result = 1;  //返回结果
-  string Other = 2;
+  string other = 2;     //信息id
 }
 
 message Result{
@@ -27,7 +27,7 @@ message Result{
   string textContent = 2;   //文本内容
   string textUrl = 3;      //文本URL
   string filePath = 4;  //文件路径
-  string errorState = 5;   //状态{200:解析成功，300:下载失败，400:解析失败}
+  string errorState = 5;   //错误状态{0:文件下载失败，1：文件解析失败}
 }
 
 

jy_publishing/task.go

@@ -9,6 +9,7 @@ import (
 	"gopkg.in/mgo.v2/bson"
 	pb "jy_publishing/proto"
 	"net"
+	"reflect"
 	"strconv"
 	"strings"
 	"time"
@@ -62,6 +63,7 @@ func taskInfo(obj interface{}) {
 // @Description 敏感词处理(title, content, attachment)
 // @Author J 2022/4/11 9:36 AM
 func Sensitive(info map[string]interface{}) {
+	util.Debug(reflect.TypeOf(info["appendInfo"]))
 	tmp := info["appendInfo"].(map[string]interface{})
 	tArr := WordsIdentify(util.ObjToString(tmp["title"]))
 	dArr := WordsIdentify(util.ObjToString(tmp["detail"]))
@@ -99,7 +101,8 @@ func Sensitive(info map[string]interface{}) {
 			Title:   tArr,
 			Detail:  dArr,
 		}
-		JyRpcSensitive(req)
+		fmt.Println(req)
+		//JyRpcSensitive(req)
 	}
 
 	//atts := tmp["attachment"].(map[string]interface{})
@@ -264,18 +267,33 @@ func InfoPub(info map[string]interface{}) {
 
 }
 
+type AttsResponse struct {
+	Other  AttsOther     `json:"other"`
+	Result []*AttsResult `json:"result"`
+}
+
+type AttsOther struct {
+	Id      string   `json:"id"`
+	Action  string   `json:"action"`
+	MsgType string   `json:"msgType"`
+	Detail  []string `json:"detail"`
+	Title   []string `json:"title"`
+}
+
+type AttsResult struct {
+	FileName    string `json:"fileName"`
+	TextUrl     string `json:"textUrl"`
+	TextContent string `json:"textContent"`
+	FilePath    string `json:"filePath"`
+	ErrorState  string `json:"errorState"`
+}
+
 // @Description 附件处理完成队列
 // @Author J 2022/4/13 3:29 PM
-func taskAtts(obj interface{}) {
-	result := obj.(*pb.FileResponse)
-	util.Info(result.String())
-	otherMap := make(map[string]interface{})
-	if err := json.Unmarshal([]byte(result.Other), &otherMap); err != nil {
-		log.Error("data Unmarshal Failed：", log.Field("error", err))
-	}
+func taskAtts(obj *AttsResponse) {
 	atts := make(map[string]interface{})
 	atts_text := make(map[string]interface{})
-	for i, r := range result.Result {
+	for i, r := range obj.Result {
 		at := make(map[string]interface{})
 		text := make(map[string]interface{})
 		at["state"] = r.ErrorState
@@ -293,15 +311,16 @@ func taskAtts(obj interface{}) {
 	attsTextJson, _ := json.Marshal(atts_text)
 	// 直接调用剑鱼接口
 	req := &pb.SensitiveRequest{
-		Id:          util.ObjToString(otherMap["id"]),
-		MsgType:     util.ObjToString(otherMap["msgType"]),
-		Action:      util.ObjToString(otherMap["action"]),
-		Title:       strings.Split(util.ObjToString(otherMap["title"]), ","),
-		Detail:      strings.Split(util.ObjToString(otherMap["detail"]), ","),
+		Id:          util.ObjToString(obj.Other.Id),
+		MsgType:     util.ObjToString(obj.Other.MsgType),
+		Action:      util.ObjToString(obj.Other.Action),
+		Title:       obj.Other.Title,
+		Detail:      obj.Other.Detail,
 		Attachments: string(attsJson),
 		AttachTxt:   string(attsTextJson),
 	}
-	JyRpcSensitive(req)
+	util.Debug(req)
+	//JyRpcSensitive(req)
 }
 
 // @Description 敏感词识别完成调用剑鱼接口