bidding_es.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859
  1. package main
  2. import (
  3. "encoding/json"
  4. "esindex/config"
  5. "esindex/oss"
  6. "fmt"
  7. "github.com/spf13/viper"
  8. "go.mongodb.org/mongo-driver/bson"
  9. "go.uber.org/zap"
  10. util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  11. "jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
  12. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  13. "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
  14. "reflect"
  15. "regexp"
  16. "strconv"
  17. "strings"
  18. "sync"
  19. "time"
  20. "unicode/utf8"
  21. )
  22. var (
  23. TimeV1 = regexp.MustCompile("(\\d{4})[年.]?$")
  24. TimeV2 = regexp.MustCompile("(\\d{4}[年.\\-/]?)(\\d{1,2}[月.\\-/]?$)")
  25. TimeClear = regexp.MustCompile("[年|月|/|.|-]")
  26. filterSpace = regexp.MustCompile("<[^>]*?>|[\\s\u3000\u2003\u00a0]")
  27. date1 = regexp.MustCompile("20[0-2][0-9][年|\\-/.][0-9]{1,2}[月|\\-/.][0-9]{1,2}[日]?")
  28. HtmlReg = regexp.MustCompile("<[^>]+>")
  29. )
  30. func biddingTask(mapInfo map[string]interface{}) {
  31. defer util.Catch()
  32. stype := util.ObjToString(mapInfo["stype"])
  33. if stype == "bidding" {
  34. uq := bson.M{"gtid": bson.M{"$gte": util.ObjToString(mapInfo["gtid"])},
  35. "lteid": bson.M{"$lte": util.ObjToString(mapInfo["lteid"])}}
  36. MgoB.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess": 8, "updatetime": time.Now().Unix()}}, false, true)
  37. }
  38. q, _ := mapInfo["query"].(map[string]interface{})
  39. if q == nil {
  40. q = map[string]interface{}{
  41. "_id": map[string]interface{}{
  42. "$gt": mongodb.StringTOBsonId(mapInfo["gtid"].(string)),
  43. "$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
  44. },
  45. }
  46. } else {
  47. //针对gte/lte,单独转换
  48. q = convertToMongoID(q)
  49. }
  50. ch := make(chan bool, 10)
  51. wg := &sync.WaitGroup{}
  52. //bidding库
  53. biddingConn := MgoB.GetMgoConn()
  54. count, _ := biddingConn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(&q).Count()
  55. log.Info("bidding表", zap.Int64("同步总数:", count))
  56. it := biddingConn.DB(config.Conf.DB.MongoB.Dbname).C(config.Conf.DB.MongoB.Coll).Find(&q).Select(map[string]interface{}{
  57. "contenthtml": 0,
  58. }).Iter()
  59. c1, index := 0, 0
  60. var indexLock sync.Mutex
  61. for tmp := make(map[string]interface{}); it.Next(tmp); c1++ {
  62. if c1%1000 == 0 {
  63. log.Info("biddingTask", zap.Int("current:", c1))
  64. log.Info("biddingTask", zap.Any("current:_id =>", tmp["_id"]))
  65. }
  66. ch <- true
  67. wg.Add(1)
  68. go func(tmp map[string]interface{}) {
  69. defer func() {
  70. <-ch
  71. wg.Done()
  72. }()
  73. if sensitive := util.ObjToString(tmp["sensitive"]); sensitive == "测试" { //bidding中有敏感词,不生索引
  74. tmp = make(map[string]interface{})
  75. return
  76. }
  77. //只针对增量数据处理;全量数据 需要用extracttype字段判断
  78. //7: 重复数据
  79. //8: 不重复
  80. if util.IntAll(tmp["dataprocess"]) != 8 {
  81. return
  82. }
  83. //// 增量数据使用上面判断;全量数据使用下面配置
  84. //-1:重复 ,1:不重复 ,0:入库 9:分类
  85. //if util.IntAll(tmp["extracttype"]) != 1 {
  86. // return
  87. //}
  88. //针对产权数据,暂时不入es 索引库
  89. if util.IntAll(tmp["infoformat"]) == 3 {
  90. return
  91. }
  92. /**
  93. 数据抽取时,有的数据的发布时间是之前的,属于增量历史数据,在判重和同步到bidding表是,会添加history_updatetime
  94. 字段,所以下面判断才会处理
  95. */
  96. if stype == "bidding_history" && tmp["history_updatetime"] == nil {
  97. return
  98. }
  99. indexLock.Lock()
  100. index++
  101. indexLock.Unlock()
  102. newTmp, update := GetEsField(tmp, stype)
  103. newTmp["dataweight"] = 0 //索引数据新增 jy置顶字段
  104. //针对中国政府采购网,单独处理
  105. if util.ObjToString(tmp["site"]) == "中国政府采购网" {
  106. objectType := MatchService(tmp)
  107. if objectType != "" {
  108. newTmp["object_type"] = objectType
  109. }
  110. }
  111. if len(update) > 0 {
  112. updateBiddingPool <- []map[string]interface{}{{
  113. "_id": tmp["_id"],
  114. },
  115. {"$set": update},
  116. }
  117. }
  118. if util.ObjToString(newTmp["spidercode"]) == "a_jyxxfbpt_gg" {
  119. // 剑鱼信息发布数据 通过udp通知信息发布程序
  120. go UdpMethod(mongodb.BsonIdToSId(newTmp["_id"]))
  121. }
  122. saveEsPool <- newTmp
  123. }(tmp)
  124. tmp = map[string]interface{}{}
  125. }
  126. wg.Wait()
  127. log.Info("biddingTask over", zap.Int("count", c1), zap.Int("index", index))
  128. ////发送udp,附件补采 才需要
  129. //data := map[string]interface{}{
  130. // "stype": "update",
  131. // "gtid": mongodb.StringTOBsonId(mapInfo["gtid"].(string)),
  132. // "lteid": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
  133. //}
  134. //target := &net.UDPAddr{
  135. // Port: 1782,
  136. // IP: net.ParseIP("127.0.0.1"),
  137. //}
  138. //bytes, _ := json.Marshal(data)
  139. //err := UdpClient.WriteUdp(bytes, udp.OP_TYPE_DATA, target)
  140. //if err != nil {
  141. // log.Info("biddingTask ", zap.Any("WriteUdp err", err), zap.Any("target", target))
  142. //}
  143. //
  144. //log.Info("biddingTask ", zap.Any("target", target), zap.Any("data", data))
  145. //
  146. //
  147. //重采平台需要
  148. //mapInfo["stype"] = ""
  149. //datas, _ := json.Marshal(mapInfo)
  150. //var next = &net.UDPAddr{
  151. // IP: net.ParseIP("127.0.0.1"),
  152. // Port: 1910,
  153. //}
  154. //log.Info("bidding index es over", zap.Any("es", next), zap.String("mapinfo", string(datas)))
  155. }
  156. //biddingAllTask 补充存量数据
  157. func biddingAllTask(mapInfo map[string]interface{}) {
  158. defer util.Catch()
  159. stype := util.ObjToString(mapInfo["stype"])
  160. q, _ := mapInfo["query"].(map[string]interface{})
  161. if q == nil {
  162. q = map[string]interface{}{
  163. "_id": map[string]interface{}{
  164. "$gt": mongodb.StringTOBsonId(mapInfo["gtid"].(string)),
  165. "$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
  166. },
  167. }
  168. } else {
  169. //针对gte/lte,单独转换
  170. q = convertToMongoID(q)
  171. }
  172. ch := make(chan bool, 50)
  173. wg := &sync.WaitGroup{}
  174. //bidding库
  175. biddingConn := MgoB.GetMgoConn()
  176. it := biddingConn.DB(config.Conf.DB.MongoB.Dbname).C(config.Conf.DB.MongoB.Coll).Find(&q).Select(map[string]interface{}{
  177. "contenthtml": 0,
  178. }).Iter()
  179. c1, index := 0, 0
  180. var indexLock sync.Mutex
  181. for tmp := make(map[string]interface{}); it.Next(tmp); c1++ {
  182. if c1%20000 == 0 {
  183. log.Info("biddingAllTask", zap.Int("current:", c1))
  184. log.Info("biddingAllTask", zap.Any("current:_id =>", tmp["_id"]))
  185. }
  186. ch <- true
  187. wg.Add(1)
  188. go func(tmp map[string]interface{}) {
  189. defer func() {
  190. <-ch
  191. wg.Done()
  192. }()
  193. if sensitive := util.ObjToString(tmp["sensitive"]); sensitive == "测试" { //bidding中有敏感词,不生索引
  194. tmp = make(map[string]interface{})
  195. return
  196. }
  197. // 针对存量数据,重复数据不进索引
  198. if util.IntAll(tmp["extracttype"]) == -1 {
  199. return
  200. }
  201. //针对产权数据,暂时不入es 索引库
  202. if util.IntAll(tmp["infoformat"]) == 3 {
  203. return
  204. }
  205. indexLock.Lock()
  206. index++
  207. indexLock.Unlock()
  208. newTmp, update := GetEsField(tmp, stype)
  209. //针对中国政府采购网,单独处理
  210. if util.ObjToString(tmp["site"]) == "中国政府采购网" {
  211. objectType := MatchService(tmp)
  212. if objectType != "" {
  213. newTmp["object_type"] = objectType
  214. }
  215. }
  216. newTmp["dataweight"] = 0 //索引数据新增 jy置顶字段
  217. if len(update) > 0 {
  218. updateBiddingPool <- []map[string]interface{}{{
  219. "_id": tmp["_id"],
  220. },
  221. {"$set": update},
  222. }
  223. }
  224. saveEsPool <- newTmp
  225. }(tmp)
  226. tmp = map[string]interface{}{}
  227. }
  228. wg.Wait()
  229. log.Info("biddingAllTask over", zap.Int("count", c1), zap.Int("index", index))
  230. }
  231. //biddingAllDataTask 处理配置文件的存量数据
  232. func biddingAllDataTask() {
  233. type Biddingall struct {
  234. Coll string
  235. Gtid string
  236. Lteid string
  237. }
  238. type RoutinesConf struct {
  239. Num int
  240. }
  241. type AllConf struct {
  242. All map[string]Biddingall
  243. Routines RoutinesConf
  244. }
  245. var all AllConf
  246. viper.SetConfigFile("biddingall.toml")
  247. viper.SetConfigName("biddingall") // 配置文件名称(无扩展名)
  248. viper.SetConfigType("toml") // 如果配置文件的名称中没有扩展名,则需要配置此项
  249. viper.AddConfigPath("./")
  250. err := viper.ReadInConfig() // 查找并读取配置文件
  251. if err != nil { // 处理读取配置文件的错误
  252. fmt.Println("ReadInConfig err =>", err)
  253. return
  254. }
  255. err = viper.Unmarshal(&all)
  256. if err != nil {
  257. fmt.Println("biddingAllDataTask Unmarshal err =>", err)
  258. return
  259. }
  260. for k, conf := range all.All {
  261. go dealData(conf.Coll, conf.Gtid, conf.Lteid, k, all.Routines.Num)
  262. }
  263. }
  264. func dealData(coll, gtid, lteid, kword string, routines int) {
  265. ch := make(chan bool, routines)
  266. wg := &sync.WaitGroup{}
  267. q := map[string]interface{}{
  268. "_id": map[string]interface{}{
  269. "$gt": mongodb.StringTOBsonId(gtid),
  270. "$lte": mongodb.StringTOBsonId(lteid),
  271. },
  272. }
  273. biddingConn := MgoB.GetMgoConn()
  274. it := biddingConn.DB(config.Conf.DB.MongoB.Dbname).C(coll).Find(&q).Select(map[string]interface{}{
  275. "contenthtml": 0,
  276. }).Iter()
  277. c1, index := 0, 0
  278. var indexLock sync.Mutex
  279. for tmp := make(map[string]interface{}); it.Next(tmp); c1++ {
  280. if c1%20000 == 0 {
  281. log.Info(kword, zap.Int("current:", c1))
  282. log.Info(kword, zap.Any("current:_id =>", tmp["_id"]))
  283. }
  284. ch <- true
  285. wg.Add(1)
  286. go func(tmp map[string]interface{}) {
  287. defer func() {
  288. <-ch
  289. wg.Done()
  290. }()
  291. if sensitive := util.ObjToString(tmp["sensitive"]); sensitive == "测试" { //bidding中有敏感词,不生索引
  292. tmp = make(map[string]interface{})
  293. return
  294. }
  295. // 针对存量数据,重复数据不进索引
  296. if util.IntAll(tmp["extracttype"]) == -1 {
  297. return
  298. }
  299. //针对产权数据,暂时不入es 索引库
  300. if util.IntAll(tmp["infoformat"]) == 3 {
  301. return
  302. }
  303. indexLock.Lock()
  304. index++
  305. indexLock.Unlock()
  306. newTmp, update := GetEsField(tmp, "biddingall")
  307. //针对中国政府采购网,单独处理
  308. if util.ObjToString(tmp["site"]) == "中国政府采购网" {
  309. objectType := MatchService(tmp)
  310. if objectType != "" {
  311. newTmp["object_type"] = objectType
  312. }
  313. }
  314. newTmp["dataweight"] = 0 //索引数据新增 jy置顶字段
  315. if len(update) > 0 {
  316. updateBiddingPool <- []map[string]interface{}{{
  317. "_id": tmp["_id"],
  318. },
  319. {"$set": update},
  320. }
  321. }
  322. saveEsPool <- newTmp
  323. }(tmp)
  324. tmp = map[string]interface{}{}
  325. }
  326. wg.Wait()
  327. log.Info(fmt.Sprintf("%s over", kword), zap.Int("count", c1), zap.Int("index", index))
  328. }
  329. func biddingTaskById(mapInfo map[string]interface{}) {
  330. defer util.Catch()
  331. stype := util.ObjToString(mapInfo["stype"])
  332. infoid := util.ObjToString(mapInfo["infoid"])
  333. tmp, _ := MgoB.FindById(config.Conf.DB.MongoB.Coll, infoid, map[string]interface{}{"contenthtml": 0})
  334. if sensitive := util.ObjToString((*tmp)["sensitive"]); sensitive == "测试" { //bidding中有敏感词,不生索引
  335. return
  336. }
  337. if util.IntAll((*tmp)["extracttype"]) == 1 {
  338. newTmp, update := GetEsField(*tmp, stype)
  339. newTmp["dataweight"] = 0 //索引数据新增 jy置顶字段
  340. if len(update) > 0 {
  341. //updateBiddingPool <- []map[string]interface{}{{
  342. // "_id": mongodb.StringTOBsonId(infoid),
  343. //},
  344. // {"$set": update},
  345. //}
  346. }
  347. saveEsPool <- newTmp
  348. }
  349. log.Info("biddingTaskById over", zap.Any("mapInfo", mapInfo))
  350. }
  351. // GetEsField @Description ES字段
  352. // @Author J 2022/6/7 11:34 AM
  353. func GetEsField(tmp map[string]interface{}, stype string) (map[string]interface{}, map[string]interface{}) {
  354. newTmp := make(map[string]interface{})
  355. update := make(map[string]interface{}) // bidding 修改字段
  356. saveErr := make(map[string]interface{})
  357. //for field, ftype := range config.Conf.DB.Es.FieldEs {
  358. for field, ftype := range BiddingField {
  359. if tmp[field] != nil { //
  360. if field == "purchasinglist" { //标的物处理
  361. purchasinglist_new := []map[string]interface{}{}
  362. if pcl, _ := tmp[field].([]interface{}); len(pcl) > 0 {
  363. for _, ls := range pcl {
  364. lsm_new := make(map[string]interface{})
  365. lsm := ls.(map[string]interface{})
  366. for pf, pftype := range BiddingLevelField[field] {
  367. lsmv := lsm[pf]
  368. if lsmv != nil && reflect.TypeOf(lsmv).String() == pftype {
  369. lsm_new[pf] = lsm[pf]
  370. }
  371. }
  372. if lsm_new != nil && len(lsm_new) > 0 {
  373. purchasinglist_new = append(purchasinglist_new, lsm_new)
  374. }
  375. }
  376. }
  377. if len(purchasinglist_new) > 0 {
  378. newTmp[field] = purchasinglist_new
  379. }
  380. } else if field == "procurementlist" {
  381. if tmp["procurementlist"] != nil {
  382. var arr []interface{}
  383. plist := tmp["procurementlist"].([]interface{})
  384. for _, p := range plist {
  385. p1 := p.(map[string]interface{})
  386. p2 := make(map[string]interface{})
  387. for k, v := range BiddingLevelField[field] {
  388. if k == "projectname" && util.ObjToString(p1[k]) == "" {
  389. p2[k] = util.ObjToString(tmp["projectname"])
  390. } else if k == "buyer" && util.ObjToString(p1[k]) == "" && util.ObjToString(tmp["buyer"]) != "" {
  391. p2[k] = util.ObjToString(tmp["buyer"])
  392. } else if k == "expurasingtime" && util.ObjToString(p1[k]) != "" {
  393. res := getMethod(util.ObjToString(p1[k]))
  394. if res != 0 {
  395. p2[k] = res
  396. }
  397. } else if p1[k] != nil && reflect.TypeOf(p1[k]).String() == v {
  398. p2[k] = p1[k]
  399. }
  400. }
  401. arr = append(arr, p2)
  402. }
  403. if len(arr) > 0 {
  404. newTmp[field] = arr
  405. }
  406. }
  407. } else if field == "projectscope" {
  408. ps, _ := tmp["projectscope"].(string)
  409. newTmp["projectscope"] = ps
  410. //新版本已无需记录长度
  411. //if len(ps) > pscopeLength {
  412. // saveErr["projectscope"] = ps
  413. // saveErr["projectscope_length"] = len(ps)
  414. //}
  415. } else if field == "winnerorder" { //中标候选
  416. winnerorder_new := []map[string]interface{}{}
  417. if winnerorder, _ := tmp[field].([]interface{}); len(winnerorder) > 0 {
  418. for _, win := range winnerorder {
  419. winMap_new := make(map[string]interface{})
  420. winMap := win.(map[string]interface{})
  421. for wf, wftype := range BiddingLevelField[field] {
  422. wfv := winMap[wf]
  423. if wfv != nil && reflect.TypeOf(wfv).String() == wftype {
  424. if wf == "sort" && util.Int64All(wfv) > 100 {
  425. continue
  426. }
  427. winMap_new[wf] = winMap[wf]
  428. }
  429. }
  430. if winMap_new != nil && len(winMap_new) > 0 {
  431. winnerorder_new = append(winnerorder_new, winMap_new)
  432. }
  433. }
  434. }
  435. if len(winnerorder_new) > 0 {
  436. newTmp[field] = winnerorder_new
  437. }
  438. } else if field == "qualifies" {
  439. //项目资质
  440. qs := []string{}
  441. if q, _ := tmp[field].([]interface{}); len(q) > 0 {
  442. for _, v := range q {
  443. v1 := v.(map[string]interface{})
  444. qs = append(qs, util.ObjToString(v1["key"]))
  445. }
  446. }
  447. if len(qs) > 0 {
  448. newTmp[field] = strings.Join(qs, ",")
  449. }
  450. } else if field == "bidopentime" {
  451. if tmp[field] != nil && tmp["bidendtime"] == nil {
  452. newTmp["bidendtime"] = tmp[field]
  453. newTmp[field] = tmp[field]
  454. } else if tmp[field] == nil && tmp["bidendtime"] != nil {
  455. newTmp["bidendtime"] = tmp[field]
  456. newTmp[field] = tmp["bidendtime"]
  457. } else {
  458. if tmp["bidopentime"] != nil {
  459. newTmp[field] = tmp["bidopentime"]
  460. }
  461. }
  462. } else if field == "detail" { //过滤
  463. detail, _ := tmp[field].(string)
  464. detail = filterSpace.ReplaceAllString(detail, "")
  465. // 不需要再保存记录长度
  466. //if len(detail) > pscopeLength {
  467. // saveErr["detail"] = detail
  468. // saveErr["detail_length"] = len(detail)
  469. //}
  470. if tmp["cleartag"] != nil {
  471. if tmp["cleartag"].(bool) {
  472. text, _ := FilterDetail(detail)
  473. newTmp[field] = util.ObjToString(tmp["title"]) + " " + text
  474. } else {
  475. newTmp[field] = util.ObjToString(tmp["title"]) + " " + detail
  476. }
  477. } else {
  478. text, b := FilterDetail(detail)
  479. newTmp[field] = util.ObjToString(tmp["title"]) + " " + text
  480. update["cleartag"] = b
  481. }
  482. } else if field == "topscopeclass" || field == "entidlist" {
  483. newTmp[field] = tmp[field]
  484. } else if field == "_id" {
  485. newTmp["_id"] = mongodb.BsonIdToSId(tmp["_id"])
  486. newTmp["id"] = mongodb.BsonIdToSId(tmp["_id"])
  487. } else if field == "publishtime" || field == "comeintime" {
  488. //字段类型不正确,特别处理
  489. if tmp[field] != nil && util.Int64All(tmp[field]) > 0 {
  490. newTmp[field] = util.Int64All(tmp[field])
  491. }
  492. } else if field == "package" {
  493. //分包信息处理
  494. packages := dealPackage(tmp)
  495. if len(packages) > 0 {
  496. newTmp["package"] = packages
  497. newTmp["subpackage"] = 1
  498. }
  499. } else { //其它字段判断数据类型,不正确舍弃
  500. if fieldval := tmp[field]; reflect.TypeOf(fieldval).String() != ftype {
  501. continue
  502. } else {
  503. if fieldval != "" {
  504. newTmp[field] = fieldval
  505. }
  506. }
  507. }
  508. }
  509. }
  510. // 附件内容长度不做限制,大于20万字符,只做记录
  511. filetext := getFileText(tmp)
  512. if len([]rune(filetext)) > 10 {
  513. newTmp["filetext"] = filetext
  514. if len([]rune(filetext)) > fileLength {
  515. saveErr["filetext"] = filetext
  516. saveErr["filetext_length"] = len([]rune(filetext))
  517. }
  518. }
  519. YuceEndtime(newTmp) // 预测结果时间
  520. if stype == "bidding" || stype == "bidding_history" || stype == "index-by-id" {
  521. newTmp["createtime"] = time.Now().Unix() // es库数据创建时间,只有增量数据有
  522. newTmp["pici"] = time.Now().Unix() //createtime跟pici一样,为了剑鱼功能需要,并行存在一段时间,之后可以删掉createtime
  523. update["pici"] = time.Now().Unix()
  524. }
  525. if len(saveErr) > 0 {
  526. saveErr["infoid"] = mongodb.BsonIdToSId(tmp["_id"])
  527. saveErr["time"] = time.Now().Unix()
  528. saveErrBidPool <- saveErr
  529. }
  530. return newTmp, update
  531. }
  532. // @Description 采购意向 预计采购时间处理
  533. // @Author J 2022/6/7 8:04 PM
  534. func getMethod(str string) int64 {
  535. if TimeV1.MatchString(str) {
  536. arr := TimeV1.FindStringSubmatch(str)
  537. st := arr[1] + "0000"
  538. parseInt, err := strconv.ParseInt(st, 10, 64)
  539. if err == nil {
  540. return parseInt
  541. }
  542. } else if TimeV2.MatchString(str) {
  543. arr := TimeV2.FindStringSubmatch(str)
  544. str1 := arr[2]
  545. if len(str1) == 1 {
  546. str1 = "0" + str1
  547. }
  548. str2 := TimeClear.ReplaceAllString(arr[1], "") + TimeClear.ReplaceAllString(str1, "") + "00"
  549. parseInt, err := strconv.ParseInt(str2, 10, 64)
  550. if err == nil {
  551. return parseInt
  552. }
  553. }
  554. return 0
  555. }
  556. func FilterDetail(text string) (string, bool) {
  557. b := false // 清理标记
  558. for _, s := range config.Conf.DB.Es.DetailFilter {
  559. reg := regexp.MustCompile(s)
  560. if reg.MatchString(text) {
  561. text = reg.ReplaceAllString(text, "")
  562. if !b {
  563. b = true
  564. }
  565. }
  566. }
  567. return text, b
  568. }
  569. // @Description 附件内容
  570. // @Author J 2022/6/7 1:54 PM
  571. func getFileText(tmp map[string]interface{}) (filetext string) {
  572. if attchMap, ok := tmp["attach_text"].(map[string]interface{}); attchMap != nil && ok {
  573. for _, tmpData1 := range attchMap {
  574. if tmpData2, ok := tmpData1.(map[string]interface{}); tmpData2 != nil && ok {
  575. for _, result := range tmpData2 {
  576. if resultMap, ok := result.(map[string]interface{}); resultMap != nil && ok {
  577. if attach_url := util.ObjToString(resultMap["attach_url"]); attach_url != "" {
  578. bs := oss.OssGetObject(attach_url, mongodb.BsonIdToSId(tmp["_id"])) //oss读数据
  579. //附件总长度限制550000,其中最后一个文件长度限制50000
  580. size := config.Conf.DB.Oss.Filesize
  581. if size <= 0 {
  582. size = 500000
  583. }
  584. if utf8.RuneCountInString(filetext+bs) < 50000+size {
  585. filetext += bs + "\n"
  586. } else {
  587. if len(bs) > 50000 {
  588. filetext += bs[0:50000]
  589. } else {
  590. filetext += bs
  591. }
  592. }
  593. //附件总长度限制550000
  594. if utf8.RuneCountInString(filetext) >= 50000+size {
  595. return
  596. }
  597. //正式环境
  598. //if utf8.RuneCountInString(filetext+bs) < fileLength {
  599. // filetext += bs + "\n"
  600. //} else {
  601. // if utf8.RuneCountInString(bs) > fileLength {
  602. // filetext = bs[0:fileLength]
  603. // } else {
  604. // filetext = bs
  605. // }
  606. // break
  607. //}
  608. }
  609. }
  610. }
  611. }
  612. }
  613. }
  614. return
  615. }
  616. // 预测结果时间
  617. func YuceEndtime(tmp map[string]interface{}) {
  618. flag := false
  619. flag2 := false
  620. scope := []string{"信息技术_运维服务", "信息技术_软件开发", "信息技术_系统集成及安全", "信息技术_其他"}
  621. titles := []string{"短信服务", "短信发送服务"}
  622. details := []string{"短信发送服务", "短信服务平台", "短信服务项目"}
  623. subscopeclass := util.ObjToString(tmp["s_subscopeclass"])
  624. //先判断满足 s_subscopeclass 条件
  625. for _, v := range scope {
  626. if strings.Contains(subscopeclass, v) {
  627. flag = true
  628. break
  629. }
  630. }
  631. //满足 s_subscopeclass ,再去判断title detail
  632. if flag {
  633. title := util.ObjToString(tmp["title"])
  634. for _, v := range titles {
  635. if strings.Contains(title, v) {
  636. flag2 = true
  637. }
  638. }
  639. if !flag2 {
  640. detail := util.ObjToString(tmp["detail"])
  641. for _, v := range details {
  642. if strings.Contains(detail, v) {
  643. flag2 = true
  644. }
  645. }
  646. }
  647. }
  648. if !flag2 {
  649. return
  650. }
  651. subtype := util.ObjToString(tmp["subtype"])
  652. if subtype == "成交" || subtype == "合同" {
  653. // yucestarttime、yuceendtime
  654. yucestarttime, yuceendtime := int64(0), int64(0)
  655. // 项目周期中
  656. if util.ObjToString(tmp["projectperiod"]) != "" {
  657. dateStr := date1.FindStringSubmatch(util.ObjToString(tmp["projectperiod"]))
  658. if len(dateStr) == 2 {
  659. sdate := FormatDateStr(dateStr[0])
  660. edate := FormatDateStr(dateStr[1])
  661. if sdate < edate && sdate != 0 && edate != 0 {
  662. yucestarttime = sdate
  663. yuceendtime = edate
  664. }
  665. }
  666. }
  667. if yucestarttime > 0 && yuceendtime > yucestarttime {
  668. tmp["yuceendtime"] = yuceendtime
  669. return
  670. }
  671. // 预测开始时间 合同签订日期
  672. if yucestarttime == 0 {
  673. if util.IntAll(tmp["signaturedate"]) <= 0 {
  674. if util.IntAll(tmp["publishtime"]) <= 0 {
  675. return
  676. } else {
  677. yucestarttime = util.Int64All(tmp["publishtime"])
  678. }
  679. } else {
  680. yucestarttime = util.Int64All(tmp["signaturedate"])
  681. }
  682. }
  683. // 预测结束时间
  684. if yucestarttime > 0 && yuceendtime == 0 {
  685. if util.IntAll(tmp["project_duration"]) > 0 && util.ObjToString(tmp["project_timeunit"]) != "" {
  686. yuceendtime = YcEndTime(yucestarttime, util.IntAll(tmp["project_duration"]), util.ObjToString(tmp["project_timeunit"]))
  687. tmp["yuceendtime"] = yuceendtime
  688. }
  689. }
  690. }
  691. }
  692. func FormatDateStr(ds string) int64 {
  693. ds = strings.Replace(ds, "年", "-", -1)
  694. ds = strings.Replace(ds, "月", "-", -1)
  695. ds = strings.Replace(ds, "日", "", -1)
  696. ds = strings.Replace(ds, "/", "-", -1)
  697. ds = strings.Replace(ds, ".", "-", -1)
  698. location, err := time.ParseInLocation(util.Date_Short_Layout, ds, time.Local)
  699. if err != nil {
  700. log.Error("FormatDateStr", zap.Error(err))
  701. return 0
  702. } else {
  703. return location.Unix()
  704. }
  705. }
  706. func YcEndTime(starttime int64, num int, unit string) int64 {
  707. yuceendtime := int64(0)
  708. if unit == "日历天" || unit == "天" || unit == "日" {
  709. yuceendtime = starttime + int64(num*86400)
  710. } else if unit == "周" {
  711. yuceendtime = time.Unix(starttime, 0).AddDate(0, 0, num*7).Unix()
  712. } else if unit == "月" {
  713. yuceendtime = time.Unix(starttime, 0).AddDate(0, num, 0).Unix()
  714. } else if unit == "年" {
  715. yuceendtime = time.Unix(starttime, 0).AddDate(num, 0, 0).Unix()
  716. } else if unit == "工作日" {
  717. n := num / 7 * 2
  718. yuceendtime = time.Unix(starttime, 0).AddDate(0, 0, num+n).Unix()
  719. }
  720. return yuceendtime
  721. }
  722. // UdpMethod @Description rpc调用信息发布程序接口
  723. // @Author J 2022/4/13 9:13 AM
  724. func UdpMethod(id string) {
  725. mapinfo := map[string]interface{}{
  726. "infoid": id,
  727. "stype": "jyfb_data_over",
  728. }
  729. datas, _ := json.Marshal(mapinfo)
  730. log.Info("UdpMethod", zap.Any("JyUdpAddr", JyUdpAddr), zap.String("mapinfo", string(datas)))
  731. _ = UdpClient.WriteUdp(datas, udp.OP_TYPE_DATA, JyUdpAddr)
  732. }
  733. //MatchService 针对中国招标网,匹配关键词打标签,object_type,货物、服务、工程,jsondata.item
  734. func MatchService(tmp map[string]interface{}) (res string) {
  735. if jsondata, ok := tmp["jsondata"]; ok {
  736. if da, ok := jsondata.(map[string]interface{}); ok {
  737. if item, ok := da["item"]; ok {
  738. services := []string{"货物", "服务", "工程"}
  739. for _, v := range services {
  740. if strings.Contains(util.ObjToString(item), v) {
  741. return v
  742. }
  743. }
  744. }
  745. }
  746. }
  747. return
  748. }
  749. //dealPackage 处理package 字段
  750. func dealPackage(tmp map[string]interface{}) (newpackages []map[string]interface{}) {
  751. package1, ok1 := tmp["package"]
  752. s_winner, ok2 := tmp["s_winner"]
  753. bidamount, ok3 := tmp["bidamount"]
  754. var innerWinners = make([]string, 0)
  755. var biaoAmounts = make([]float64, 0)
  756. // 三个字段都存在
  757. if ok3 && ok2 && ok1 {
  758. packageMap, ok := package1.(map[string]interface{})
  759. if ok {
  760. if len(packageMap) >= 2 {
  761. var packages = make([]map[string]interface{}, 0)
  762. //var newTmp = make(map[string]interface{})
  763. winner_amount_count := 0
  764. for _, pack := range packageMap {
  765. var newPackage = make(map[string]interface{})
  766. pac, okk := pack.(map[string]interface{})
  767. if okk {
  768. _, okk1 := pac["winner"]
  769. _, okk2 := pac["bidamount"]
  770. _, okk3 := pac["name"]
  771. if okk1 {
  772. innerWinners = append(innerWinners, util.ObjToString(pac["winner"]))
  773. }
  774. if okk2 {
  775. biaoAmounts = append(biaoAmounts, util.Float64All(pac["bidamount"]))
  776. }
  777. //winner bidamount 二个字段都存在
  778. if okk1 && okk2 {
  779. winner_amount_count++
  780. newPackage["winner"] = pac["winner"]
  781. newPackage["bidamount"] = pac["bidamount"]
  782. if okk3 {
  783. newPackage["name"] = pac["name"]
  784. }
  785. packages = append(packages, newPackage)
  786. }
  787. }
  788. }
  789. //出现次数大于1
  790. if winner_amount_count > 1 {
  791. swinner := util.ObjToString(s_winner)
  792. swinners := strings.Split(swinner, ",")
  793. //判断里外 winner 是否相等
  794. eq := StringSliceValuesEqual(swinners, innerWinners)
  795. if eq {
  796. //判断金额相等
  797. if Float64Equal1Precision(Float64SliceSum(biaoAmounts), util.Float64All(bidamount)) {
  798. newpackages = packages
  799. }
  800. }
  801. }
  802. }
  803. }
  804. }
  805. return
  806. }