|
|
@@ -8,6 +8,7 @@ import (
|
|
|
"encoding/json"
|
|
|
"esindex/config"
|
|
|
"esindex/oss"
|
|
|
+ "fmt"
|
|
|
"go.mongodb.org/mongo-driver/bson"
|
|
|
"go.uber.org/zap"
|
|
|
"reflect"
|
|
|
@@ -61,6 +62,7 @@ func biddingTask(mapInfo map[string]interface{}) {
|
|
|
for tmp := make(map[string]interface{}); it.Next(tmp); c1++ {
|
|
|
if c1%1000 == 0 {
|
|
|
log.Info("biddingTask", zap.Int("current:", c1))
|
|
|
+ log.Info("biddingAllTask", zap.Any("current:_id =>", tmp["_id"]))
|
|
|
}
|
|
|
ch <- true
|
|
|
wg.Add(1)
|
|
|
@@ -73,12 +75,24 @@ func biddingTask(mapInfo map[string]interface{}) {
|
|
|
tmp = make(map[string]interface{})
|
|
|
return
|
|
|
}
|
|
|
+ //只针对增量数据处理；全量数据 需要用extracttype字段判断
|
|
|
if util.IntAll(tmp["dataprocess"]) != 8 {
|
|
|
return
|
|
|
}
|
|
|
+ //// 增量数据使用上面判断；全量数据使用下面配置
|
|
|
//if util.IntAll(tmp["extracttype"]) != 1 {
|
|
|
// return
|
|
|
//}
|
|
|
+
|
|
|
+ //针对产权数据，暂时不入es 索引库
|
|
|
+ if util.IntAll(tmp["infoformat"]) == 3 {
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ 数据抽取时，有的数据的发布时间是之前的，属于增量历史数据，在判重和同步到bidding表是，会添加history_updatetime
|
|
|
+ 字段，所以下面判断才会处理
|
|
|
+ */
|
|
|
if stype == "bidding_history" && tmp["history_updatetime"] == nil {
|
|
|
return
|
|
|
}
|
|
|
@@ -87,6 +101,15 @@ func biddingTask(mapInfo map[string]interface{}) {
|
|
|
indexLock.Unlock()
|
|
|
newTmp, update := GetEsField(tmp, stype)
|
|
|
newTmp["dataweight"] = 0 //索引数据新增 jy置顶字段
|
|
|
+
|
|
|
+ //针对中国政府采购网,单独处理
|
|
|
+ if util.ObjToString(tmp["site"]) == "中国政府采购网" {
|
|
|
+ objectType := MatchService(tmp)
|
|
|
+ if objectType != "" {
|
|
|
+ newTmp["object_type"] = objectType
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
if len(update) > 0 {
|
|
|
updateBiddingPool <- []map[string]interface{}{{
|
|
|
"_id": tmp["_id"],
|
|
|
@@ -136,12 +159,13 @@ func biddingAllTask(mapInfo map[string]interface{}) {
|
|
|
biddingConn := MgoB.GetMgoConn()
|
|
|
it := biddingConn.DB(config.Conf.DB.MongoB.Dbname).C(config.Conf.DB.MongoB.Coll).Find(&q).Select(map[string]interface{}{
|
|
|
"contenthtml": 0,
|
|
|
- }).Sort("-_id").Iter()
|
|
|
+ }).Iter()
|
|
|
c1, index := 0, 0
|
|
|
var indexLock sync.Mutex
|
|
|
for tmp := make(map[string]interface{}); it.Next(tmp); c1++ {
|
|
|
if c1%20000 == 0 {
|
|
|
log.Info("biddingAllTask", zap.Int("current:", c1))
|
|
|
+ log.Info("biddingAllTask", zap.Any("current:_id =>", tmp["_id"]))
|
|
|
}
|
|
|
ch <- true
|
|
|
wg.Add(1)
|
|
|
@@ -154,9 +178,16 @@ func biddingAllTask(mapInfo map[string]interface{}) {
|
|
|
tmp = make(map[string]interface{})
|
|
|
return
|
|
|
}
|
|
|
+ // 针对17833，需要单独屏蔽这个判断，不需要处理
|
|
|
if util.IntAll(tmp["extracttype"]) == -1 {
|
|
|
return
|
|
|
}
|
|
|
+
|
|
|
+ //针对产权数据，暂时不入es 索引库
|
|
|
+ if util.IntAll(tmp["infoformat"]) == 3 {
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
indexLock.Lock()
|
|
|
index++
|
|
|
indexLock.Unlock()
|
|
|
@@ -547,3 +578,22 @@ func UdpMethod(id string) {
|
|
|
log.Info("UdpMethod", zap.Any("JyUdpAddr", JyUdpAddr), zap.String("mapinfo", string(datas)))
|
|
|
_ = UdpClient.WriteUdp(datas, udp.OP_TYPE_DATA, JyUdpAddr)
|
|
|
}
|
|
|
+
|
|
|
+//MatchService 针对中国招标网，匹配关键词打标签,object_type,货物、服务、工程，jsondata.item
|
|
|
+func MatchService(tmp map[string]interface{}) (res string) {
|
|
|
+ if jsondata, ok := tmp["jsondata"]; ok {
|
|
|
+ if da, ok := jsondata.(map[string]interface{}); ok {
|
|
|
+ if item, ok := da["item"]; ok {
|
|
|
+ fmt.Println("item=>", item)
|
|
|
+ services := []string{"货物", "服务", "工程"}
|
|
|
+ for _, v := range services {
|
|
|
+ if strings.Contains(util.ObjToString(item), v) {
|
|
|
+ return v
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return
|
|
|
+}
|
wcc