Merge branch 'dev0.3' of http://192.168.3.207:10080/qmx/datatag into dev0.3

src/config.json

@@ -1,6 +1,6 @@
 {
   "port": "7000",
-  "udpport": ":11116",
+  "udpport": ":11117",
   "mgodb": "192.168.3.207:27092",
   "dbsize": 5,
   "dbname": "datatag",

src/service/tag_rule_service.go

@@ -175,7 +175,7 @@ func (r *TagRule) TagSetup() {
 	defer qu.Catch()
 	if r.Method() == "POST" {
 		_id := r.GetString("_id")
-		i_isuse := r.GetString("i_isuse")
+		i_isuse, _ := r.GetInteger("i_isuse")
 		set := bson.M{
 			"$set": bson.M{
 				"i_isuse": i_isuse,

src/service/task_rule.go

@@ -130,7 +130,6 @@ func (task *TaskRule) TaskStart() {
 		IP:   net.ParseIP(data["addr"].(string)),
 		Port: qu.IntAll(data["port"]),
 	}
-
 	err := Udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
 	if err != nil {
 		task.ServeJson(map[string]interface{}{

src/tagservice/src/config.json

@@ -1,12 +1,19 @@
 {
-	"addr": "192.168.3.207:27092",
-	"db": "mxs",
-	"c": "test2",
-	"size": 10,
-	"es":{
-		"index": "bidding_v2",
-        "type": "bidding",
-		"addr": "http://192.168.3.11:9800",
-        "pool": 12
-	}
+	"mgodb": "192.168.3.207:27092",
+    "dbsize": 10,
+    "dbname": "datatag",
+	"udpport": ":11116",
+	"esport": 1479,
+	"esaddr": "127.0.0.1",
+	"fields": {
+		"title": 1,
+		"detail": 1,
+		"tagname": 1
+	},
+	"history": {
+		"tasktype": "1",
+		"tagmgo": "127.0.0.1:27092",
+		"tagdb": "mxs",
+		"tagcoll": "test2"
+	}	
 }

src/tagservice/src/data.go

@@ -1,176 +0,0 @@
-package main
-
-import (
-	"log"
-	"os"
-	qu "qfw/util"
-	"regexp"
-	"strings"
-
-	"github.com/tealeg/xlsx"
-)
-
-func init() {
-	InitTags() //读取规则表
-
-}
-
-//匹配方式map
-var export_matchtype = map[string]interface{}{
-	"标题匹配":   "title",
-	"全文匹配":   "detail",
-	"标的物匹配":  "purchasing",
-	"项目名称匹配": "projectname",
-	"采购单位匹配": "buyer",
-	"中标单位匹配": "s_winner",
-	"附件匹配":   "filetext",
-	// "8":      "采购单位匹配(正则)",
-	// "9":      "中标单位匹配(正则)",
-}
-var LetterCase = regexp.MustCompile("[A-Za-z]")
-var Rules []*Rule
-
-//一条规则
-type Rule struct {
-	KW      *KeyWord
-	AW      *AddWord
-	TagName string
-}
-
-//关键词类型
-type KeyWord struct {
-	KeyReg     []*regexp.Regexp
-	MatchType  []string     //关键词的匹配方式
-	KeyWordMap map[int]bool //记录KeyReg中字母规则
-}
-
-//附加词类型
-type AddWord struct {
-	KeyReg     []*regexp.Regexp
-	MatchType  []string     //附加词的匹配方式
-	AddWordMap map[int]bool //记录KeyReg中字母规则
-}
-
-func InitTags() {
-	defer qu.Catch()
-	xf, err := xlsx.OpenFile("tag.xlsx")
-	if err != nil {
-		log.Println("fields file not foud", err.Error())
-		os.Exit(0)
-	}
-	sheet := xf.Sheets[0]
-	for i, row := range sheet.Rows {
-		/*
-			注：从表格中取出的内容可能为空，split截取后数组长度为1
-		*/
-		if i == 0 {
-			continue
-		}
-
-		if len(row.Cells) < 5 { //表格有问题
-			log.Println("excel row length is wrong.	row:", i+1)
-			os.Exit(0)
-		}
-		//标签名称
-		tagname := row.Cells[4].Value
-		if tagname == "" {
-			log.Println("excel tagname is empty. row:", i+1)
-			os.Exit(0)
-		}
-		//附加词匹配方式
-		awm := row.Cells[3].Value
-		awmArr := []string{}
-		for _, av := range strings.Split(awm, ",") {
-			if field := qu.ObjToString(export_matchtype[av]); field != "" {
-				awmArr = append(awmArr, field)
-			}
-		}
-
-		//附加词
-		tmp_aw := []*AddWord{}
-		addword := row.Cells[2].Value
-		aw_commaArr := strings.Split(addword, ",")
-		for _, comma := range aw_commaArr {
-			aw := &AddWord{}
-			aw.AddWordMap = make(map[int]bool)
-			aw.MatchType = awmArr
-			aw_addArr := strings.Split(comma, "&&")
-			if len(aw_addArr) == 1 { //,
-				tmp_aw := aw_addArr[0]
-				if tmp_aw != "" {
-					if LetterCase.MatchString(tmp_aw) {
-						tmp_aw = strings.ToUpper(tmp_aw) //附加词中有英文全部转为大写
-						aw.AddWordMap[len(aw.KeyReg)] = true
-					}
-					aw.KeyReg = append(aw.KeyReg, regexp.MustCompile(tmp_aw))
-				}
-			} else { //&&
-				for _, and := range aw_addArr {
-					if and != "" {
-						if LetterCase.MatchString(and) {
-							and = strings.ToUpper(and) //附加词中有英文全部转为大写
-							aw.AddWordMap[len(aw.KeyReg)] = true
-						}
-						aw.KeyReg = append(aw.KeyReg, regexp.MustCompile(and))
-					}
-				}
-			}
-			tmp_aw = append(tmp_aw, aw)
-		}
-
-		//关键词匹配方式
-		kwm := row.Cells[1].Value
-		kwmArr := []string{}
-		for _, kv := range strings.Split(kwm, ",") {
-			if field := qu.ObjToString(export_matchtype[kv]); field != "" {
-				kwmArr = append(kwmArr, field)
-			}
-		}
-
-		//关键词
-		tmp_kw := []*KeyWord{}
-		keyword := row.Cells[0].Value
-		kw_commaArr := strings.Split(keyword, ",")
-		for _, comma := range kw_commaArr {
-			kw := &KeyWord{}
-			kw.KeyWordMap = make(map[int]bool)
-			kw.MatchType = kwmArr
-			kw_addArr := strings.Split(comma, "&&")
-			if len(kw_addArr) == 1 { //,
-				tmp_kw := kw_addArr[0]
-				if tmp_kw != "" {
-					if LetterCase.MatchString(tmp_kw) {
-						tmp_kw = strings.ToUpper(tmp_kw)
-						kw.KeyWordMap[len(kw.KeyReg)] = true
-					}
-					kw.KeyReg = append(kw.KeyReg, regexp.MustCompile(tmp_kw))
-				}
-			} else { //&&
-				for _, and := range kw_addArr {
-					if and != "" {
-						if LetterCase.MatchString(and) {
-							and = strings.ToUpper(and)
-							kw.KeyWordMap[len(kw.KeyReg)] = true
-						}
-						kw.KeyReg = append(kw.KeyReg, regexp.MustCompile(and))
-					}
-				}
-			}
-			tmp_kw = append(tmp_kw, kw)
-		}
-
-		//组合
-		for _, tk := range tmp_kw {
-			for _, aw := range tmp_aw {
-				rule := &Rule{}
-				rule.KW = tk
-				rule.AW = aw
-				rule.TagName = tagname
-				Rules = append(Rules, rule)
-			}
-		}
-	}
-	for i, r := range Rules {
-		qu.Debug(i, r.TagName, r.KW.KeyReg, len(r.KW.KeyReg), r.KW.MatchType, len(r.KW.MatchType), r.KW.KeyWordMap, "---", r.AW.KeyReg, len(r.AW.KeyReg), r.AW.MatchType, len(r.AW.MatchType), r.AW.AddWordMap)
-	}
-}

src/tagservice/src/historytask.go

@@ -0,0 +1,4 @@
+//mgo 跑历史数据
+package main
+
+func test() {}

src/util/task.go → src/tagservice/src/newtask.go

@@ -1,4 +1,4 @@
-package util
+package main
 
 import (
 	"log"
@@ -7,7 +7,10 @@ import (
 	"qfw/util/elastic"
 	"regexp"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
+	//"gopkg.in/mgo.v2/bson"
 )
 
 //匹配方式map
@@ -21,22 +24,30 @@ var task_export_matchtype = map[string]interface{}{
 	"7": "s_winner",
 }
 var LetterCase = regexp.MustCompile("[A-Za-z]")
+var FilteReg = regexp.MustCompile("[()（）{}]*")
+var TaskList = make(map[string]*Task) //存储启动任务
+var Fields = map[string]interface{}{"title": 1, "detail": 1, "tagname": 1}
 
 //任务模型
 type Task struct {
 	//任务信息
-	Id      string      //任务id
-	StartId string      //起始id
-	From    string      //数据出处(es mongodb)
-	To      string      //数据更新去处(es mongodb)
-	Index   string      //es index
-	Itype   string      //es type
-	MgoColl string      //mgo coll
-	Rules   []*Tag_Rule //任务相关规则(对数据打标签)
-	IsRun   bool        //是否运行
+	Id      string //任务id
+	StartId string //起始id
+	From    string //数据出处(es mongodb)
+	//To      string  //数据更新去处(es mongodb)
+	Index   string  //es index
+	Itype   string  //es type
+	MgoDb   string  //mgo db
+	MgoColl string  //mgo coll
+	Rules   []*Rule //任务相关规则(对数据打标签)
+	IsRun   bool    //是否运行
+	IsIndex bool    //是否同步es
 	//存储相关
+	Wg             *sync.WaitGroup
+	Lock           *sync.Mutex
 	Mgo            *mongo.MongodbSim             //mgo
 	Es             *elastic.Elastic              //es
+	DataChan       chan bool                     //
 	EsUpdateCache  chan map[string]string        //es更新集合
 	MgoUpdataCache chan []map[string]interface{} //mgo更新集合
 	SP             chan bool                     //批量更新时的线程控制
@@ -44,7 +55,7 @@ type Task struct {
 }
 
 //规则
-type Tag_Rule struct {
+type Rule struct {
 	KW      *KeyWord
 	AW      *AddWord
 	TagName string
@@ -64,113 +75,194 @@ type AddWord struct {
 	AddWordMap map[int]bool //记录KeyReg中字母规则
 }
 
-//更新es
-func (t *Task) UpdateEs() {
-	log.Println("Es Save...")
-	arru := make([]map[string]string, 200)
-	indexu := 0
-	for {
-		select {
-		case v := <-t.EsUpdateCache:
-			arru[indexu] = v
-			indexu++
-			if indexu == 200 {
-				t.SP <- true
-				go func(arru []map[string]string) {
-					defer func() {
-						<-t.SP
-					}()
-					elastic.BulkUpdateArr(t.Index, t.Itype, arru)
-				}(arru)
-				arru = make([]map[string]string, 200)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				t.SP <- true
-				go func(arru []map[string]string) {
-					defer func() {
-						<-t.SP
-					}()
-					elastic.BulkUpdateArr(t.Index, t.Itype, arru)
-				}(arru[:indexu])
-				arru = make([]map[string]string, 200)
-				indexu = 0
-			}
-		}
+func StartTask(taskid string) {
+	t := &Task{}
+	t.InitTask(taskid)
+	t.IsRun = true //更新任务状态
+	qu.Debug(t.Id, t.From, t.Index, t.Itype, t.IsRun, t.IsIndex)
+	TaskList[taskid] = t //加入map
+	if t.From == "mongodb" {
+		t.RunMgo() //增量
+		//go t.UpdateMgo() //开启mgo保存
+	} else {
+		t.RunEs()
+		go t.UpdateEs() //开启es保存
 	}
 }
 
-//更新mongo
-func (t *Task) UpdateMgo() {
-	log.Println("Mgo Save...")
-	arru := make([][]map[string]interface{}, 200)
-	indexu := 0
-	for {
-		select {
-		case v := <-t.MgoUpdataCache:
-			arru[indexu] = v
-			indexu++
-			if indexu == 200 {
-				t.SP <- true
-				go func(arru [][]map[string]interface{}) {
-					defer func() {
-						<-t.SP
-					}()
-					t.Mgo.UpdateBulk(t.MgoColl, arru...)
-				}(arru)
-				arru = make([][]map[string]interface{}, 200)
-				indexu = 0
+//mgo增量
+func (t *Task) RunMgo() {
+	if TaskList[t.Id] == nil {
+		return
+	}
+	//oldId := t.StartId //记录起始id
+	queryOne := map[string]interface{}{
+		"_id": map[string]interface{}{
+			"$gte": qu.StringTOBsonId(t.StartId),
+		},
+	}
+	data, _ := t.Mgo.Find(t.MgoColl, queryOne, `{"_id":-1}`, nil, false, 0, 1) //找最后一条数据
+	endId := qu.BsonIdToSId((*data)[0]["_id"])
+	if endId <= t.StartId { //判断id
+		return
+	}
+	sid := t.StartId
+	query := map[string]interface{}{
+		"_id": map[string]interface{}{
+			"$gt":  qu.StringTOBsonId(sid),
+			"$lte": qu.StringTOBsonId(endId),
+		},
+	}
+	sess := t.Mgo.GetMgoConn()
+	defer t.Mgo.DestoryMongoConn(sess)
+	count, _ := sess.DB(t.MgoDb).C(t.MgoColl).Find(&query).Count()
+	log.Println("查询语句:", query, "查询总数:", count)
+	it := sess.DB(t.MgoDb).C(t.MgoColl).Find(&query).Select(Fields).Sort("_id").Iter()
+
+	t.Wg = &sync.WaitGroup{}
+	t.Lock = &sync.Mutex{}
+	t.DataChan = make(chan bool, 20)
+	update := [][]map[string]interface{}{}
+	//遍历
+	index := 0
+	n := int64(0)
+	for tmp := map[string]interface{}{}; it.Next(&tmp); index++ {
+		if index%500 == 0 {
+			log.Println("current:", index)
+		}
+		tid := qu.BsonIdToSId(tmp["_id"])
+		t.Wg.Add(1)
+		t.DataChan <- true
+		go func(tmp map[string]interface{}) {
+			defer func() {
+				<-t.DataChan
+				t.Wg.Done()
+			}()
+			tmpTagName := map[string]bool{}
+			for _, r := range t.Rules {
+				//关键词匹配
+			L:
+				for _, kwm := range r.KW.MatchType {
+					if text := qu.ObjToString(tmp[kwm]); text != "" {
+						text = ProcessData(text)
+						for _, kw_reg := range r.KW.KeyReg {
+							if kw_reg.MatchString(text) { //关键词匹配成功
+								//qu.Debug(kwm, kw_reg)
+								//kwMatch = true
+								//关键词匹配成功后，匹配附加词
+								//qu.Debug(len(r.AW.KeyReg), r.AW.KeyReg)
+								if len(r.AW.KeyReg) == 0 { //无附加词
+									//adMatch = true
+									tmpTagName[r.TagName] = true
+									break L
+								} else {
+									for _, awm := range r.AW.MatchType {
+										if text := qu.ObjToString(tmp[awm]); text != "" {
+											for _, aw_reg := range r.AW.KeyReg {
+												if aw_reg.MatchString(text) { //附加词匹配成功
+													//adMatch = true
+													tmpTagName[r.TagName] = true
+													break L
+												}
+											}
+										}
+									}
+								}
+							}
+						}
+					}
+				}
 			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				t.SP <- true
-				go func(arru [][]map[string]interface{}) {
-					defer func() {
-						<-t.SP
-					}()
-					t.Mgo.UpdateBulk(t.MgoColl, arru...)
-				}(arru[:indexu])
-				arru = make([][]map[string]interface{}, 200)
-				indexu = 0
+			if len(tmpTagName) > 0 {
+				tagname_arr := []string{}
+				for tagname, _ := range tmpTagName {
+					tagname_arr = append(tagname_arr, tagname)
+				}
+
+				atomic.AddInt64(&n, +1) //n++
+
+				t.Lock.Lock()
+				idAndSet := []map[string]interface{}{}
+				_id := map[string]interface{}{
+					"_id": tmp["_id"],
+				}
+				set := map[string]interface{}{
+					"$set": map[string]interface{}{
+						"tagname": strings.Join(tagname_arr, ","),
+					},
+				}
+				idAndSet = append(idAndSet, _id) //第一个为查询条件
+				idAndSet = append(idAndSet, set) //第二个为更新内容
+
+				update = append(update, idAndSet)
+				if len(update) > 500 {
+					t.Mgo.UpdateBulk(t.MgoColl, update...)
+					update = [][]map[string]interface{}{} //更新后把数据置空
+				}
+				t.Lock.Unlock()
 			}
+		}(tmp)
+		if tid > t.StartId {
+			t.StartId = tid
 		}
+		tmp = map[string]interface{}{}
+	}
+	t.Wg.Wait()
+	t.Lock.Lock()
+	if len(update) > 0 {
+		t.Mgo.UpdateBulk(t.MgoColl, update...)
+		update = [][]map[string]interface{}{} //更新后把数据置空
 	}
+	t.Lock.Unlock()
+	log.Println("Update Count:", n)
+	//更新起始id
+	setid := map[string]interface{}{
+		"$set": map[string]interface{}{
+			"s_startid": t.StartId,
+		},
+	}
+	go Mgo.Update("taskinfo", `{"_id":"`+t.Id+`"}`, setid, false, false)
+	//是否同步es
+	if t.IsIndex {
+		go processEs(sid, endId)
+	}
+	time.AfterFunc(time.Minute*5, t.RunMgo)
+}
+
+func (t *Task) RunEs() {
+
 }
 
 //初始化任务信息
-func InitTask(taskid string) {
-	t := &Task{}
+func (t *Task) InitTask(taskid string) {
 	data, _ := Mgo.FindById("taskinfo", taskid, nil)
 	t.Id = taskid
 	t.StartId = qu.ObjToString((*data)["s_startid"])
 	from := qu.ObjToString((*data)["s_fromtype"])
 	t.From = from
-	to := qu.ObjToString((*data)["s_totype"])
-	t.To = to
-	if from == to { //同库
-		url := qu.ObjToString((*data)["s_fromdburl"])
-		dbname := qu.ObjToString((*data)["s_fromdbname"])
-		coll := qu.ObjToString((*data)["s_fromdbcoll"])
-		if from == "es" { //es
-			t.InitEs(url, dbname, coll)
-		} else { //mgo
-			t.InitMgo(url, dbname, coll)
+	url := qu.ObjToString((*data)["s_fromdburl"])
+	dbname := qu.ObjToString((*data)["s_fromdbname"])
+	coll := qu.ObjToString((*data)["s_fromdbcoll"])
+	if from == "mongodb" { //初始化mgo
+		t.Mgo = &mongo.MongodbSim{
+			MongodbAddr: url,
+			Size:        10,
+			DbName:      dbname,
 		}
-	} else { //异库
-		fromdburl := qu.ObjToString((*data)["s_fromdburl"])
-		fromdbname := qu.ObjToString((*data)["s_fromdbname"])
-		fromdbcoll := qu.ObjToString((*data)["s_fromdbcoll"])
-		todburl := qu.ObjToString((*data)["s_todburl"])
-		todbname := qu.ObjToString((*data)["s_todbname"])
-		todbcoll := qu.ObjToString((*data)["s_todbcoll"])
-		if from == "es" {
-			t.InitEs(fromdburl, fromdbname, fromdbcoll)
-			t.InitMgo(todburl, todbname, todbcoll)
-		} else {
-			t.InitMgo(fromdburl, fromdbname, fromdbcoll)
-			t.InitEs(todburl, todbname, todbcoll)
+		t.Mgo.InitPool()
+		t.MgoDb = dbname
+		t.MgoColl = coll
+		if s_synces := qu.ObjToString((*data)["s_synces"]); s_synces == "1" {
+			t.IsIndex = true //mgo打标签是否同步es
 		}
+	} else { //初始化es
+		t.Es = &elastic.Elastic{
+			S_esurl: url,
+			I_size:  15,
+		}
+		t.Es.InitElasticSize()
+		t.Index = dbname
+		t.Itype = coll
 	}
 	t.EsUpdateCache = make(chan map[string]string, 500)
 	t.MgoUpdataCache = make(chan []map[string]interface{}, 500)
@@ -273,7 +365,7 @@ func (t *Task) InitRules(tasktype string) {
 			//组合
 			for _, tk := range tmp_kw {
 				for _, aw := range tmp_aw {
-					rule := &Tag_Rule{}
+					rule := &Rule{}
 					rule.KW = tk
 					rule.AW = aw
 					rule.TagName = tagname
@@ -284,30 +376,88 @@ func (t *Task) InitRules(tasktype string) {
 		}
 	}
 
-	// for i, r := range t.Rules {
-	// 	qu.Debug(i, r.TagName, r.KW.KeyReg, len(r.KW.KeyReg), r.KW.MatchType, len(r.KW.MatchType), r.KW.KeyWordMap, "---", r.AW.KeyReg, len(r.AW.KeyReg), r.AW.MatchType, len(r.AW.MatchType), r.AW.AddWordMap)
-	// }
-	// qu.Debug(t.Id, t.From, t.To, t.StartId)
+	for i, r := range t.Rules {
+		qu.Debug(i, r.TagName, r.KW.KeyReg, len(r.KW.KeyReg), r.KW.MatchType, len(r.KW.MatchType), r.KW.KeyWordMap, "---", r.AW.KeyReg, len(r.AW.KeyReg), r.AW.MatchType, len(r.AW.MatchType), r.AW.AddWordMap)
+	}
 }
 
-//初始化mgo
-func (t *Task) InitMgo(url, dbname, coll string) {
-	t.Mgo = &mongo.MongodbSim{
-		MongodbAddr: url,
-		Size:        10,
-		DbName:      dbname,
+//处理文本
+func ProcessData(text string) string {
+	text = strings.ToUpper(text)               //文本中的英文全转为大写
+	text = FilteReg.ReplaceAllString(text, "") //去除一些特殊符号
+	return text
+}
+
+//更新es
+func (t *Task) UpdateEs() {
+	log.Println("Es Save...")
+	arru := make([]map[string]string, 200)
+	indexu := 0
+	for {
+		select {
+		case v := <-t.EsUpdateCache:
+			arru[indexu] = v
+			indexu++
+			if indexu == 200 {
+				t.SP <- true
+				go func(arru []map[string]string) {
+					defer func() {
+						<-t.SP
+					}()
+					elastic.BulkUpdateArr(t.Index, t.Itype, arru)
+				}(arru)
+				arru = make([]map[string]string, 200)
+				indexu = 0
+			}
+		case <-time.After(1000 * time.Millisecond):
+			if indexu > 0 {
+				t.SP <- true
+				go func(arru []map[string]string) {
+					defer func() {
+						<-t.SP
+					}()
+					elastic.BulkUpdateArr(t.Index, t.Itype, arru)
+				}(arru[:indexu])
+				arru = make([]map[string]string, 200)
+				indexu = 0
+			}
+		}
 	}
-	t.Mgo.InitPool()
-	t.MgoColl = coll
 }
 
-//初始化es
-func (t *Task) InitEs(url, dbname, coll string) {
-	t.Es = &elastic.Elastic{
-		S_esurl: url,
-		I_size:  15,
+//更新mongo
+func (t *Task) UpdateMgo() {
+	log.Println("Mgo Save...")
+	arru := make([][]map[string]interface{}, 200)
+	indexu := 0
+	for {
+		select {
+		case v := <-t.MgoUpdataCache:
+			arru[indexu] = v
+			indexu++
+			if indexu == 200 {
+				t.SP <- true
+				go func(arru [][]map[string]interface{}) {
+					defer func() {
+						<-t.SP
+					}()
+					t.Mgo.UpdateBulk(t.MgoColl, arru...)
+				}(arru)
+				arru = make([][]map[string]interface{}, 200)
+				indexu = 0
+			}
+		case <-time.After(1000 * time.Millisecond):
+			if indexu > 0 {
+				t.SP <- true
+				go func(arru [][]map[string]interface{}) {
+					defer func() {
+						<-t.SP
+					}()
+					t.Mgo.UpdateBulk(t.MgoColl, arru...)
+				}(arru[:indexu])
+				arru = make([][]map[string]interface{}, 200)
+				indexu = 0
+			}
+		}
 	}
-	t.Es.InitElasticSize()
-	t.Index = dbname
-	t.Itype = coll
 }

src/tagservice/src/udptask.go

@@ -0,0 +1,65 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	mu "mfw/util"
+	"net"
+	qu "qfw/util"
+)
+
+var Udpclient mu.UdpClient //udp对象
+
+func init() {
+	qu.Debug("222")
+	updport := Sysconfig["udpport"].(string)
+	Udpclient = mu.UdpClient{Local: updport, BufSize: 1024}
+	log.Println("Udp服务监听", updport)
+	Udpclient.Listen(processUdpMsg)
+}
+
+func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
+	switch act {
+	case mu.OP_TYPE_DATA:
+		qu.Debug("-------------")
+		var mapInfo map[string]interface{}
+		err := json.Unmarshal(data, &mapInfo)
+		if err != nil {
+			go Udpclient.WriteUdp([]byte{}, mu.OP_NOOP, ra) //回应上一个节点
+		} else {
+			taskid := qu.ObjToString(mapInfo["taskid"])
+			stype := qu.ObjToString(mapInfo["stype"])
+			by, _ := json.Marshal(map[string]interface{}{
+				"taskid": taskid,
+				"stype":  stype,
+			})
+			go Udpclient.WriteUdp(by, mu.OP_NOOP, ra) //回应上一个节点
+			if stype == "startTask" {
+				if TaskList[taskid] == nil { //启动
+					go StartTask(taskid)
+				}
+			} else if stype == "stopTask" {
+				if TaskList[taskid] != nil { //关闭
+
+				}
+			}
+		}
+	case mu.OP_NOOP: //下个节点回应
+
+	}
+
+}
+
+func processEs(sid, eid string) {
+	by, _ := json.Marshal(map[string]interface{}{
+		"gtid":  sid,
+		"lteid": eid,
+		"stype": "bidding",
+	})
+	log.Println("Mgo打标签完毕同步Es：", qu.ObjToString(Sysconfig["esaddr"]), string(by))
+	addr := &net.UDPAddr{
+		IP:   net.ParseIP(qu.ObjToString(Sysconfig["esaddr"])),
+		Port: qu.IntAll(Sysconfig["esport"]),
+	}
+	Udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
+}

src/web/templates/login.html

@@ -93,7 +93,7 @@ function login(){
 		data:{"email":email,"pwd":pwd},
 		success:function(r){
 			if(r.checked){
-				window.location.href="/service/task/list"
+				window.location.href="/service/rule/list"
 			}else{
 				alert("fail");
 			}

src/web/templates/private/tag_rule_create.html

@@ -401,6 +401,7 @@
         dataMap.s_tagfield = $('#fieldSelect').val()
         dataMap.s_tasktype = $('#taskSelect').val()
         dataMap.s_query = $('#query').val()
+        dataMap.s_pretagfield = $('#pretagfield').val()
         if (dataMap.s_tagname == "" || dataMap.s_tagfield == "" || dataMap.s_tasktype == "") {
             alert("请填写必须字段！")
             return
@@ -535,9 +536,9 @@
             key_flag = 1;
             var index = selectIndex[0] - 1;
             $('#modal-add-keyword').modal("show");
-            $('#keyword_txt').val(o_rules[index].s_matchkey);
-            $('#addword_txt').val(o_rules[index].s_addkey);
-            $('#notword_txt').val(o_rules[index].s_notkey);
+            $('#keyword_txt').val(o_list[index].s_matchkey);
+            $('#addword_txt').val(o_list[index].s_addkey);
+            $('#notword_txt').val(o_list[index].s_notkey);
         }else {
             showTip("请选中需要修改的关键词，且只能选中一条", 1000)
         }
@@ -553,12 +554,12 @@
             $('#modal-add-keyword').modal('hide');
             if (key_flag == 1) {
                 var index = selectIndex[0] - 1;
-                o_rules.splice(index, 1, rule)
+                o_list.splice(index, 1, rule)
             }else {
-                o_rules.push(rule)
+                o_list.push(rule)
             }
             $('#keywords').dataTable().fnClearTable();
-            $('#keywords').dataTable().fnAddData(o_rules);
+            $('#keywords').dataTable().fnAddData(o_list);
         }else {
             showTip("关键词不能为空", 500)
         }

src/web/templates/private/tag_rule_edit.html

@@ -6,7 +6,7 @@
 {{include "com/modal.html"}}
 <div class="content-wrapper">
     <section class="content-header">
-        <h1>新建标签规则
+        <h1>编辑标签规则
             <small>
                 <button class="btn btn-primary btn-sm" onclick="importRule()"><i class="fa fa-fw fa-cloud-upload fa-lg"></i>导入关键词</button>
                 <form style="display:none" id='uploadform' method='post'>
@@ -16,8 +16,8 @@
         </h1>
         <ol class="breadcrumb">
             <li><a href="#"><i class="fa fa-dashboard"></i> 首页</a></li>
-            <li><a href="/service/rule/list">标签规则列表</a></li>
-            <li><a href="#">新建标签规则</a></li>
+            <li><a href="/service/rule/list"> 标签规则列表</a></li>
+            <li><a href="#"> 编辑标签规则</a></li>
         </ol>
     </section>
     <!-- Main content -->
@@ -138,7 +138,7 @@
 
     var setValue = 0;                   // 4: 关键词匹配方式，6：附加词匹配方式
     var selectIndex = [];               //选中的编号
-    var o_list = [];
+    var o_list = dataMap.o_list;
     var matchCode = [];     //匹配方式code
     var matchName = [];     //匹配名字
 
@@ -190,6 +190,7 @@
                     cell.innerHTML = i + 1;
                 });
             },
+            "data": dataMap.o_list,
             "columns": [
                 {"data": null, width: "1%"},
                 {"data": "", width: "1%", render: function() {
@@ -340,6 +341,7 @@
         dataMap.s_tagfield = $('#fieldSelect').val()
         dataMap.s_tasktype = $('#taskSelect').val()
         dataMap.s_query = $('#query').val()
+        dataMap.s_pretagfield = $('#pretagfield').val()
         if (dataMap.s_tagname == "" || dataMap.s_tagfield == "" || dataMap.s_tasktype == "") {
             alert("请填写必须字段！")
             return
@@ -459,4 +461,71 @@
         matchName.length = 0;
         matchCode.length = 0;
     }
+
+    var key_flag = -1;      // 0：新增，1：修改
+    //关键词表格新增
+    function add_Rule() {
+        key_flag = 0;
+        $('#keyword_txt').val("");
+        $('#addword_txt').val("");
+        $('#notword_txt').val("");
+        $('#modal-add-keyword').modal("show");
+    }
+    //关键词修改
+    function modify_Rule(){
+        if (selectIndex.length == 1) {
+            key_flag = 1;
+            var index = selectIndex[0] - 1;
+            $('#modal-add-keyword').modal("show");
+            $('#keyword_txt').val(o_list[index].s_matchkey);
+            $('#addword_txt').val(o_list[index].s_addkey);
+            $('#notword_txt').val(o_list[index].s_notkey);
+        }else {
+            showTip("请选中需要修改的关键词，且只能选中一条", 1000)
+        }
+
+    }
+    function saveKeyword() {
+        var key = $('#keyword_txt').val();
+        var addkey = $('#addword_txt').val();
+        if (key != "") {
+            var rule = {};
+            rule["s_matchkey"] = key.replace("，", ",");
+            if (addkey != "") rule["s_addkey"] = addkey.replace("，", ",");
+            $('#modal-add-keyword').modal('hide');
+            if (key_flag == 1) {
+                var index = selectIndex[0] - 1;
+                o_list.splice(index, 1, rule)
+            }else {
+                o_list.push(rule)
+            }
+            $('#keywords').dataTable().fnClearTable();
+            $('#keywords').dataTable().fnAddData(o_list);
+        }else {
+            showTip("关键词不能为空", 500)
+        }
+    }
+    //关键词删除
+    function del_Rule() {
+        if (keyTableChecked && selectIndex.length > 0) {
+            showConfirm("确定删除？", function() {
+                var tmp = []
+                for (var i in selectIndex) {
+                    var j = selectIndex[i] - 1;
+                    tmp.push(j)
+                }
+                o_list = deleteArrayVal(o_list, tmp)
+                $('#keywords').dataTable().fnClearTable();
+                $('#keywords').dataTable().fnAddData(o_list);
+
+                keyTableChecked = false;
+                if (o_list.length == 0) {
+                    $('#allCheckbox').prop("checked", false);
+                    $('#keywords').dataTable().fnClearTable();
+                }
+            })
+        }else {
+            showTip("请选中需要删除的关键词", 500)
+        }
+    }
 </script>