task.go 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982
  1. package main
  2. import (
  3. "encoding/json"
  4. "field_sync/config"
  5. "field_sync/oss"
  6. "fmt"
  7. "net"
  8. "reflect"
  9. "regexp"
  10. "sort"
  11. "strconv"
  12. "strings"
  13. "time"
  14. "log"
  15. // "go.mongodb.org/mongo-driver/bson"
  16. util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  17. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  18. "jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
  19. "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
  20. )
  21. var (
  22. regLetter = regexp.MustCompile("[a-z]*")
  23. cityEndReg = regexp.MustCompile("(区|县|市)$")
  24. )
  25. func biddingTask(data []byte, mapInfo map[string]interface{}) {
  26. defer util.Catch()
  27. stype := util.ObjToString(mapInfo["stype"])
  28. q, _ := mapInfo["query"].(map[string]interface{})
  29. bkey, _ := mapInfo["bkey"].(string)
  30. if q == nil {
  31. q = map[string]interface{}{
  32. "_id": map[string]interface{}{
  33. "$gt": mongodb.StringTOBsonId(mapInfo["gtid"].(string)),
  34. "$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
  35. },
  36. }
  37. }
  38. //extract库
  39. extractConn := MgoE.GetMgoConn()
  40. defer MgoE.DestoryMongoConn(extractConn)
  41. extractResult := extractConn.DB(MgoE.DbName).C(config.Conf.DB.MongoE.Coll).Find(q).Select(map[string]interface{}{
  42. "field_source": 0,
  43. "kvtext": 0,
  44. }).Sort("_id").Iter()
  45. eMap := map[string]map[string]interface{}{}
  46. extCount, repeatCount := 0, 0
  47. for tmp := make(map[string]interface{}); extractResult.Next(tmp); extCount++ {
  48. if util.IntAll(tmp["repeat"]) == 1 {
  49. repeatCount++
  50. }
  51. tid := mongodb.BsonIdToSId(tmp["_id"])
  52. eMap[tid] = tmp
  53. tmp = make(map[string]interface{})
  54. }
  55. log.Println("抽取表 数据量", extCount, "重复数据量", repeatCount)
  56. //bidding库
  57. biddingConn := MgoB.GetMgoConn()
  58. count, _ := biddingConn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(&q).Count()
  59. log.Println("bidding表 同步总数:", count)
  60. c := 0
  61. if count < 500000 {
  62. var res []map[string]interface{}
  63. result := biddingConn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(q).Select(map[string]interface{}{
  64. "contenthtml": 0,
  65. }).Iter()
  66. for tmp := make(map[string]interface{}); result.Next(tmp); {
  67. res = append(res, tmp)
  68. tmp = make(map[string]interface{})
  69. }
  70. MgoB.DestoryMongoConn(biddingConn)
  71. log.Println("查询结果 bidding", count, "抽取:", extCount)
  72. c = doIndex(res, eMap, bkey, stype)
  73. } else {
  74. log.Println("查询结果 数据量太大,放弃", count)
  75. MgoB.DestoryMongoConn(biddingConn)
  76. }
  77. log.Println("bidding sync...over all", count, "extract sync ", c)
  78. // NextNode(mapInfo, stype)
  79. // NextNodePro(mapInfo, stype)
  80. // NextNodeTidb(mapInfo, stype)
  81. // if stype == "bidding_history" {
  82. // NextNodeBidData(mapInfo) // bidding-data数据
  83. // NextNodeTidbQyxy(mapInfo) // tidb-企业数据
  84. // NextNodeHn(mapInfo)
  85. // }
  86. // if stype == "bidding" {
  87. // uq := bson.M{"gtid": util.ObjToString(mapInfo["gtid"]), "lteid": util.ObjToString(mapInfo["lteid"])}
  88. // MgoBP.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess_ai": 6, "updatetime": time.Now().Unix()}}, false, true)
  89. // }
  90. //领域标签处理的数据 id段
  91. // if stype == "bidding_history" {
  92. // MgoB.Save("field_data_record", map[string]interface{}{"gtid": mapInfo["gtid"], "lteid": mapInfo["lteid"], "status": 0})
  93. // }
  94. }
  95. func biddingAllTask() {
  96. defer util.Catch()
  97. // q, _ := mapInfo["query"].(map[string]interface{})
  98. // if q == nil {
  99. q := map[string]interface{}{
  100. "site": "千里马", "comeintime": map[string]interface{}{"$gte": 1729483200},
  101. }
  102. // }
  103. //extract库
  104. // extractConn := MgoE.GetMgoConn()
  105. // defer MgoE.DestoryMongoConn(extractConn)
  106. // extractResult := extractConn.DB(MgoE.DbName).C(config.Conf.DB.MongoE.Coll1).Find(q).Select(map[string]interface{}{
  107. // "field_source": 0,
  108. // "kvtext": 0,
  109. // }).Sort("-_id").Iter()
  110. //bidding库
  111. biddingConn := MgoB.GetMgoConn()
  112. defer MgoB.DestoryMongoConn(biddingConn)
  113. count := 0
  114. // var compare map[string]interface{}
  115. result := biddingConn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(q).Select(map[string]interface{}{
  116. "contenthtml": 0,
  117. "field_source": 0,
  118. }).Sort("-_id").Iter()
  119. for tmp := make(map[string]interface{}); result.Next(tmp); count++ {
  120. update := map[string]interface{}{}
  121. del := map[string]interface{}{} //记录extract没有值而bidding中有值的字段
  122. //对比方法----------------
  123. // for {
  124. // if compare == nil {
  125. // compare = make(map[string]interface{})
  126. // if !extractResult.Next(compare) {
  127. // break
  128. // }
  129. // }
  130. tid := mongodb.BsonIdToSId(tmp["_id"])
  131. log.Println("tid ", tid)
  132. compares, _ := MgoE.FindById(config.Conf.DB.MongoE.Coll, tid, nil)
  133. if compares != nil && len(*compares) > 0 {
  134. compare := *compares
  135. // cid := mongodb.BsonIdToSId(compare["_id"])
  136. // if cid == tid {
  137. // if compare["history_updatetime"] == nil { //extract中history_updatetime不存在跳过
  138. // continue
  139. // }
  140. //更新bidding表;bidding表modifyinfo中的字段不更新
  141. modifyinfo := make(map[string]bool)
  142. if tmpmodifyinfo, ok := tmp["modifyinfo"].(map[string]interface{}); ok && tmpmodifyinfo != nil {
  143. for k := range tmpmodifyinfo {
  144. modifyinfo[k] = true
  145. }
  146. }
  147. for _, k := range config.Conf.Serve.FieldS {
  148. v1 := compare[k] //extract
  149. v2 := tmp[k] //bidding
  150. if v2 == nil && v1 != nil {
  151. update[k] = v1
  152. } else if v2 != nil && v1 != nil && !modifyinfo[k] {
  153. update[k] = v1
  154. } else if v2 != nil && v1 == nil && !modifyinfo[k] {
  155. if k == "s_subscopeclass" && del["subscopeclass"] == nil {
  156. continue
  157. } else if k == "s_topscopeclass" && del["topscopeclass"] == nil {
  158. continue
  159. }
  160. del[k] = 1
  161. //util.Debug("抽取结果没有值,bidding有值:field--", k, del)
  162. }
  163. }
  164. if util.IntAll(compare["repeat"]) == 1 {
  165. update["extracttype"] = -1
  166. update["dataprocess"] = 7
  167. if compare["repeat_id"] != nil {
  168. update["repeat_id"] = compare["repeat_id"]
  169. }
  170. } else {
  171. update["extracttype"] = 1
  172. update["dataprocess"] = 8
  173. }
  174. // break
  175. // } else {
  176. // if cid < tid {
  177. // compare = nil
  178. // continue
  179. // } else {
  180. // break
  181. // }
  182. // }
  183. //------------------对比结束
  184. //处理分类
  185. // if compare != nil { //extract
  186. fieldFun(compare, update)
  187. // compare = nil
  188. // }
  189. // entidlist
  190. // extractMap := make(map[string]interface{})
  191. if update["s_winner"] != "" {
  192. cid := companyFun(update)
  193. if len(cid) > 0 {
  194. update["entidlist"] = cid
  195. // extractMap["entidlist"] = cid
  196. }
  197. }
  198. // if len(extractMap) > 0 {
  199. // updateExtPool <- []map[string]interface{}{
  200. // {"_id": tmp["_id"]},
  201. // {"$set": extractMap},
  202. // }
  203. // }
  204. // 附件有效字段
  205. if i := validFile(tmp); i != 0 {
  206. if i == -1 {
  207. update["isValidFile"] = false
  208. } else {
  209. update["isValidFile"] = true
  210. }
  211. }
  212. if len(update) > 0 {
  213. if len(del) > 0 { //删除的字段
  214. // updateBidPool <- []map[string]interface{}{{
  215. // "_id": tmp["_id"],
  216. // },
  217. // {"$set": update, "$unset": del},
  218. // }
  219. MgoB.UpdateById(config.Conf.DB.MongoB.Coll, tmp["_id"], map[string]interface{}{"$set": update, "$unset": del})
  220. } else {
  221. // updateBidPool <- []map[string]interface{}{{
  222. // "_id": tmp["_id"],
  223. // },
  224. // {"$set": update},
  225. // }
  226. MgoB.UpdateById(config.Conf.DB.MongoB.Coll, tmp["_id"], map[string]interface{}{"$set": update})
  227. }
  228. }
  229. } else {
  230. continue
  231. }
  232. // }
  233. if count%1000 == 0 {
  234. log.Println("biddingTask current", count)
  235. }
  236. tmp = make(map[string]interface{})
  237. }
  238. log.Println("biddingAll sync...over all", count)
  239. }
  240. func doIndex(infos []map[string]interface{}, eMap map[string]map[string]interface{}, bkey, stype string) int {
  241. syncNo := 0 //抽取表数据同步数量
  242. //对比两张表数据,减少查询次数
  243. var compare map[string]interface{}
  244. var bidUpdate [][]map[string]interface{}
  245. var extUpdate [][]map[string]interface{}
  246. //SaveEsLock := &sync.Mutex{}
  247. log.Println("start ...")
  248. for n, tmp := range infos {
  249. tid := mongodb.BsonIdToSId(tmp["_id"])
  250. update := map[string]interface{}{} //要更新的mongo数据
  251. del := map[string]interface{}{}
  252. //对比方法----------------
  253. if eMap[tid] != nil {
  254. compare = eMap[tid]
  255. // if stype == "bidding" {
  256. // // 增量id段 正常数据
  257. // if dg := util.IntAll(compare["dataging"]); dg == 1 { //extract中dataging=1跳过
  258. // tmp = make(map[string]interface{})
  259. // compare = nil
  260. // continue
  261. // }
  262. // delete(eMap, tid)
  263. // }
  264. // if stype == "bidding_history" {
  265. // //增量id段 历史数据
  266. if compare["history_updatetime"] == nil { //extract中history_updatetime不存在跳过
  267. tmp = make(map[string]interface{})
  268. compare = nil
  269. continue
  270. }
  271. delete(eMap, tid)
  272. // }
  273. syncNo++
  274. log.Println("抽取区域 省", compare["area"], " 市 ", compare["city"], " 区 ", compare["district"], " id ", tid)
  275. for _, k := range config.Conf.Serve.FieldS {
  276. v1 := compare[k] //extract
  277. v2 := tmp[k] //bidding
  278. if v2 == nil && v1 != nil {
  279. update[k] = v1
  280. } else if v2 != nil && v1 != nil {
  281. update[k] = v1
  282. } else if v2 != nil && v1 == nil {
  283. if k == "s_subscopeclass" && del["subscopeclass"] == nil {
  284. continue
  285. } else if k == "s_topscopeclass" && del["topscopeclass"] == nil {
  286. continue
  287. } else if k == "city" || k == "district" {
  288. update[k] = ""
  289. } else {
  290. del[k] = 1
  291. }
  292. }
  293. }
  294. // 附件重采,数据同步时不更新判重标识
  295. if util.IntAll(compare["repeat"]) == 1 {
  296. update["extracttype"] = -1
  297. update["dataprocess"] = 7
  298. if compare["repeat_id"] != nil {
  299. update["repeat_id"] = compare["repeat_id"]
  300. }
  301. } else {
  302. update["extracttype"] = 1
  303. update["dataprocess"] = 8
  304. }
  305. } else {
  306. compare = nil
  307. if util.IntAll(tmp["dataging"]) == 1 { //修改未抽取的bidding数据的dataging
  308. update["dataging"] = 0
  309. }
  310. update["dataprocess"] = 8
  311. }
  312. //下面可以多线程跑的--->
  313. //处理分类
  314. if compare != nil { //extract
  315. fieldFun(compare, update)
  316. // publishtime 20230523
  317. if util.IntAll(tmp["publishtime"]) == -1 {
  318. if pb := methodPb(compare); pb > 0 {
  319. update["publishtime"] = pb
  320. }
  321. }
  322. compare = nil
  323. }
  324. //------------------对比结束
  325. //处理key descript
  326. if bkey == "" {
  327. DealInfo(&tmp, &update)
  328. }
  329. // entidlist
  330. extractMap := make(map[string]interface{})
  331. if update["s_winner"] != "" {
  332. cid := companyFun(update)
  333. if len(cid) > 0 {
  334. tmp["entidlist"] = cid
  335. update["entidlist"] = cid
  336. extractMap["entidlist"] = cid
  337. }
  338. }
  339. // 6.10 剑鱼发布信息分类处理, 写在这里是为了修改抽取表
  340. typeFunc(tmp, update, extractMap)
  341. if len(extractMap) > 0 {
  342. if extractMap["toptype"] != nil && extractMap["subtype"] == nil {
  343. extUpdate = append(extUpdate, []map[string]interface{}{
  344. {"_id": tmp["_id"]},
  345. {"$set": extractMap, "$unset": map[string]interface{}{"subtype": ""}},
  346. })
  347. } else {
  348. extUpdate = append(extUpdate, []map[string]interface{}{
  349. {"_id": tmp["_id"]},
  350. {"$set": extractMap},
  351. })
  352. }
  353. if len(extUpdate) >= MgoBulkSize {
  354. tmps := extUpdate
  355. MgoE.UpdateBulk(config.Conf.DB.MongoE.Coll, tmps...)
  356. extUpdate = [][]map[string]interface{}{}
  357. }
  358. }
  359. // 附件有效字段
  360. if i := validFile(tmp); i != 0 {
  361. if i == -1 {
  362. tmp["isValidFile"] = false
  363. update["isValidFile"] = false
  364. } else {
  365. tmp["isValidFile"] = true
  366. update["isValidFile"] = true
  367. }
  368. }
  369. // 2024-02-21 徐志恒 情报标签字段
  370. toptype := util.ObjToString(tmp["toptype"])
  371. subtype := util.ObjToString(tmp["subtype"])
  372. buyerclass := util.ObjToString(tmp["buyerclass"])
  373. if buyerclass != "" {
  374. update["buyer_type"] = getStr(buyerclass)
  375. }
  376. s_topscopeclass := util.ObjToString(update["s_topscopeclass"])
  377. if (tmp["tag_topinformation"] != nil && (subtype == "合同" || subtype == "中标" || subtype == "成交" || subtype == "采购意向" || toptype == "招标")) || (tmp["tag_topinformation"] == nil && toptype == "拟建" && strings.Contains(s_topscopeclass, "建筑工程")) {
  378. update["tag_set"] = getTagSet(tmp, compare)
  379. }
  380. if len(update) > 0 {
  381. log.Println("保存bidding区域 省", update["area"], " 市 ", update["city"], " 区 ", update["district"], " id ", tid)
  382. if len(del) > 0 {
  383. bidUpdate = append(bidUpdate, []map[string]interface{}{{
  384. "_id": tmp["_id"],
  385. },
  386. {"$set": update, "$unset": del},
  387. })
  388. } else {
  389. bidUpdate = append(bidUpdate, []map[string]interface{}{{
  390. "_id": tmp["_id"],
  391. },
  392. {"$set": update},
  393. })
  394. }
  395. if len(bidUpdate) >= MgoBulkSize {
  396. tmps := bidUpdate
  397. MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, tmps...)
  398. bidUpdate = [][]map[string]interface{}{}
  399. }
  400. }
  401. if n%500 == 0 {
  402. log.Println("biddingTask current ", n)
  403. }
  404. tmp = make(map[string]interface{})
  405. }
  406. //SaveEsLock.Lock()
  407. if len(bidUpdate) > 0 {
  408. tmps := bidUpdate
  409. MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, tmps...)
  410. bidUpdate = [][]map[string]interface{}{}
  411. }
  412. if len(extUpdate) > 0 {
  413. tmps := extUpdate
  414. MgoE.UpdateBulk(config.Conf.DB.MongoE.Coll, tmps...)
  415. extUpdate = [][]map[string]interface{}{}
  416. }
  417. //SaveEsLock.Unlock()
  418. return syncNo
  419. }
  420. // @Description subscopeclass、topscopeclass、package
  421. // 20230523 多包处理 subpackage = 1
  422. // @Author J 2022/6/7 5:54 PM
  423. func fieldFun(compare, update map[string]interface{}) {
  424. subscopeclass, _ := compare["subscopeclass"].([]interface{}) //subscopeclass
  425. if subscopeclass != nil {
  426. m1 := map[string]bool{}
  427. newclass := []string{}
  428. for _, sc := range subscopeclass {
  429. sclass, _ := sc.(string)
  430. if !m1[sclass] {
  431. m1[sclass] = true
  432. newclass = append(newclass, sclass)
  433. }
  434. }
  435. update["s_subscopeclass"] = strings.Join(newclass, ",")
  436. update["subscopeclass"] = newclass
  437. }
  438. topscopeclass, _ := compare["topscopeclass"].([]interface{}) //topscopeclass
  439. if topscopeclass != nil {
  440. m2 := map[string]bool{}
  441. newclass := []string{}
  442. for _, tc := range topscopeclass {
  443. tclass, _ := tc.(string)
  444. tclass = regLetter.ReplaceAllString(tclass, "") // 去除字母
  445. if !m2[tclass] {
  446. m2[tclass] = true
  447. newclass = append(newclass, tclass)
  448. }
  449. }
  450. update["topscopeclass"] = topscopeclass
  451. update["s_topscopeclass"] = strings.Join(newclass, ",")
  452. }
  453. if package1 := compare["package"]; package1 != nil {
  454. packageM, _ := package1.(map[string]interface{})
  455. update["package"] = packageM
  456. for _, p := range packageM {
  457. pm, _ := p.(map[string]interface{})
  458. if util.ObjToString(pm["winner"]) != "" || util.Float64All(pm["budget"]) > 0 ||
  459. util.Float64All(pm["bidamount"]) > 0 {
  460. update["multipackage"] = 1
  461. break
  462. }
  463. }
  464. } else {
  465. update["multipackage"] = 0
  466. }
  467. // subpackage
  468. if compare["package"] != nil && compare["s_winner"] != nil && compare["bidamount"] != nil {
  469. pg := compare["package"].(map[string]interface{})
  470. if len(pg) > 1 {
  471. var bmt []float64
  472. var swn []string
  473. for _, p := range pg {
  474. p1 := p.(map[string]interface{})
  475. if p1["bidamount"] != nil {
  476. bmt = append(bmt, util.Float64All(p1["bidamount"]))
  477. }
  478. if w := util.ObjToString(p1["winner"]); w != "" {
  479. swn = append(swn)
  480. }
  481. }
  482. if len(bmt) > 1 && len(swn) > 1 {
  483. sn := strings.Split(util.ObjToString(compare["s_winner"]), ",")
  484. sort.Strings(sn)
  485. sort.Strings(swn)
  486. swn1 := util.ObjArrToStringArr(Duplicate(swn)) // 去重
  487. if strings.Join(swn1, ",") == strings.Join(sn, ",") {
  488. bidamount := 0.0
  489. for _, f := range bmt {
  490. bidamount += f
  491. }
  492. if bidamount == util.Float64All(compare["bidamount"]) {
  493. update["subpackage"] = 1
  494. }
  495. }
  496. }
  497. }
  498. }
  499. }
  500. // @Description entidlist
  501. // @Author J 2022/6/7 2:36 PM
  502. func companyFun(tmp map[string]interface{}) (cid []string) {
  503. sWinnerarr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
  504. for _, w := range sWinnerarr {
  505. if w != "" {
  506. id := redis.GetStr("qyxy_id", w)
  507. if id == "" {
  508. ents, _ := MgoQ.Find(config.Conf.DB.MongoQ.Coll, map[string]interface{}{"company_name": w}, map[string]interface{}{"updatetime": -1}, map[string]interface{}{"company_name": 1}, false, -1, -1)
  509. if len(*ents) > 0 {
  510. id = util.ObjToString((*ents)[0]["_id"])
  511. redis.PutCKV("qyxy_id", w, id)
  512. } else {
  513. ent, _ := MgoP.FindOne(config.Conf.DB.MongoP.Coll, map[string]interface{}{"history_name": w})
  514. if len(*ent) > 0 {
  515. id = util.ObjToString((*ent)["company_id"])
  516. redis.PutCKV("qyxy_id", w, id)
  517. }
  518. }
  519. }
  520. if id == "" {
  521. id = "-"
  522. }
  523. cid = append(cid, id)
  524. }
  525. }
  526. return cid
  527. }
  528. // @Description update 修改bidding表,extractM修改抽取表
  529. // @Author J 2022/6/10 10:29 AM
  530. func typeFunc(tmp, update, extractM map[string]interface{}) {
  531. if jyData, ok := tmp["jyfb_data"].(map[string]interface{}); ok {
  532. if t := util.ObjToString(jyData["type"]); t != "" {
  533. switch t {
  534. //case "采购信息":
  535. case "招标公告":
  536. if util.ObjToString(tmp["toptype"]) != "招标" {
  537. update["toptype"] = "招标"
  538. extractM["toptype"] = "招标"
  539. delete(update, "subtype")
  540. }
  541. case "采购意向":
  542. if util.ObjToString(tmp["toptype"]) != "采购意向" {
  543. update["toptype"] = "采购意向"
  544. update["subtype"] = "采购意向"
  545. extractM["toptype"] = "采购意向"
  546. extractM["subtype"] = "采购意向"
  547. }
  548. case "招标预告":
  549. if util.ObjToString(tmp["toptype"]) != "预告" {
  550. update["toptype"] = "预告"
  551. extractM["toptype"] = "预告"
  552. delete(update, "subtype")
  553. }
  554. case "招标结果":
  555. if util.ObjToString(tmp["toptype"]) != "结果" {
  556. update["toptype"] = "结果"
  557. extractM["toptype"] = "结果"
  558. delete(update, "subtype")
  559. }
  560. }
  561. }
  562. }
  563. }
  564. // @Description 附件有效字段(isValidFile)
  565. // @Author J 2022/7/8 14:41
  566. func validFile(tmp map[string]interface{}) int {
  567. isContinue := false
  568. if pinfo, o := tmp["projectinfo"].(map[string]interface{}); o {
  569. if atts, o1 := pinfo["attachments"].(map[string]interface{}); o1 {
  570. for _, att := range atts {
  571. if att == nil {
  572. continue
  573. }
  574. if reflect.TypeOf(att).String() == "string" {
  575. continue
  576. }
  577. att1 := att.(map[string]interface{})
  578. if fid := util.ObjToString(att1["fid"]); fid != "" {
  579. isContinue = true
  580. break
  581. }
  582. }
  583. if isContinue {
  584. if attachTxt, o := tmp["attach_text"].(map[string]interface{}); o {
  585. if len(attachTxt) > 0 {
  586. for _, at := range attachTxt {
  587. at1 := at.(map[string]interface{})
  588. if len(at1) > 0 {
  589. for k, _ := range at1 {
  590. if reflect.TypeOf(at1[k]).String() == "string" {
  591. continue
  592. }
  593. at2 := at1[k].(map[string]interface{})
  594. s := strings.ToLower(util.ObjToString(at2["file_name"]))
  595. if !strings.Contains(s, "jpg") || !strings.Contains(s, "jpeg") != strings.Contains(s, "png") ||
  596. strings.Contains(s, "pdf") {
  597. if strings.Contains(s, "swf") || strings.Contains(s, "html") {
  598. return -1
  599. } else if AnalysisFile(oss.OssGetObject(util.ObjToString(at2["attach_url"]))) {
  600. return 1
  601. }
  602. }
  603. }
  604. break
  605. } else {
  606. break
  607. }
  608. }
  609. }
  610. }
  611. flag := false
  612. for _, att := range atts {
  613. if att == nil {
  614. continue
  615. }
  616. if reflect.TypeOf(att).String() == "string" {
  617. continue
  618. }
  619. att1 := att.(map[string]interface{})
  620. if fid := util.ObjToString(att1["fid"]); fid != "" {
  621. ftype := strings.ToLower(util.ObjToString(tmp["ftype"]))
  622. if ftype != "swf" && ftype != "html" && oss.OssObjExists("jy-datafile", fid) {
  623. return 1
  624. } else {
  625. flag = true
  626. }
  627. }
  628. }
  629. if flag {
  630. return -1
  631. }
  632. }
  633. }
  634. }
  635. return 0
  636. }
  637. // @Description id不变,内容变化 重新索引数据
  638. // @Author J 2022/8/10 13:29
  639. func taskinfo(id string) {
  640. tmp, _ := MgoB.FindById("bidding", id, nil)
  641. if tmp == nil || len(*tmp) == 0 {
  642. log.Println(fmt.Sprintf("taskinfo bidding id=%s 未查询到数据", id))
  643. return
  644. }
  645. extractM, _ := MgoE.FindById(config.Conf.DB.MongoE.Coll, id, nil)
  646. if extractM == nil || len(*extractM) == 0 {
  647. extractM, _ = MgoE.FindById(config.Conf.DB.MongoE.Coll1, id, nil)
  648. if extractM == nil || len(*extractM) == 0 {
  649. log.Println(fmt.Sprintf("taskinfo extract id=%s 未查询到数据", id))
  650. return
  651. }
  652. }
  653. update := map[string]interface{}{} //要更新的mongo数据
  654. //更新bidding表字段
  655. for _, k := range config.Conf.Serve.FieldS {
  656. v1 := (*extractM)[k] //extract
  657. v2 := (*tmp)[k] //bidding
  658. if v2 == nil && v1 != nil {
  659. update[k] = v1
  660. } else if v2 != nil && v1 != nil {
  661. update[k] = v1
  662. } else if v2 != nil && v1 == nil {
  663. if k == "city" || k == "district" {
  664. update[k] = ""
  665. }
  666. }
  667. }
  668. if util.IntAll((*extractM)["repeat"]) == 1 {
  669. update["extracttype"] = -1
  670. update["dataprocess"] = 7
  671. if (*extractM)["repeat_id"] != nil {
  672. update["repeat_id"] = (*extractM)["repeat_id"]
  673. }
  674. } else {
  675. update["extracttype"] = 1
  676. update["dataprocess"] = 8
  677. }
  678. //处理分类
  679. fieldFun(*extractM, update)
  680. extractMap := make(map[string]interface{})
  681. if util.ObjToString((*tmp)["s_winner"]) != "" {
  682. cid := companyFun(*tmp)
  683. if len(cid) > 0 {
  684. update["entidlist"] = cid
  685. extractMap["entidlist"] = cid
  686. }
  687. MgoE.UpdateById(config.Conf.DB.MongoE.Coll, id, map[string]interface{}{"$set": extractMap})
  688. }
  689. // 附件有效字段
  690. if i := validFile(*tmp); i != 0 {
  691. if i == -1 {
  692. update["isValidFile"] = false
  693. } else {
  694. update["isValidFile"] = true
  695. }
  696. }
  697. if len(update) > 0 {
  698. MgoB.UpdateById(config.Conf.DB.MongoB.Coll, id, map[string]interface{}{"$set": update})
  699. }
  700. mapinfo := map[string]interface{}{
  701. "infoid": id,
  702. "stype": "index-by-id",
  703. }
  704. datas, _ := json.Marshal(mapinfo)
  705. var next = &net.UDPAddr{
  706. IP: net.ParseIP(config.Conf.Udp.Next.Addr),
  707. Port: util.IntAll(config.Conf.Udp.Next.Port),
  708. }
  709. log.Println("nsq data over es ", next, " mapinfo ", string(datas))
  710. _ = UdpClient.WriteUdp(datas, udp.OP_TYPE_DATA, next)
  711. }
  712. var DateTimeSelect = []string{"bidopentime", "bidendtime", "signaturedate", "comeintime"}
  713. // @Description 发布时间处理
  714. // @Author J 2023/5/23 14:32
  715. func methodPb(tmp map[string]interface{}) int64 {
  716. if tmp["ext_publishtime"] != nil {
  717. if newPb := util.Int64All(tmp["ext_publishtime"]); newPb < time.Now().Unix() && newPb > 1420041600 {
  718. return newPb
  719. }
  720. }
  721. for _, d := range DateTimeSelect {
  722. if tmp[d] != nil && util.Int64All(tmp[d]) < time.Now().Unix() {
  723. return util.Int64All(tmp[d])
  724. }
  725. }
  726. return 0
  727. }
  728. // Duplicate
  729. // @Description 去重
  730. // @Author J 2023/5/24 09:53
  731. func Duplicate(a interface{}) (ret []interface{}) {
  732. va := reflect.ValueOf(a)
  733. for i := 0; i < va.Len(); i++ {
  734. if i > 0 && reflect.DeepEqual(va.Index(i-1).Interface(), va.Index(i).Interface()) {
  735. continue
  736. }
  737. ret = append(ret, va.Index(i).Interface())
  738. }
  739. return ret
  740. }
  741. // @Description 获取情报标签
  742. // @Author 徐志恒 2024/2/21 09:53
  743. func getTagSet(tmp, compare map[string]interface{}) map[string]map[string]interface{} {
  744. tagSet := map[string]map[string]interface{}{}
  745. wuye := map[string]interface{}{}
  746. buyer := util.ObjToString(compare["buyer"])
  747. publishtime := util.Int64All(tmp["publishtime"])
  748. bidamount := util.Float64All(compare["bidamount"])
  749. toptype := util.ObjToString(tmp["toptype"])
  750. subtype := util.ObjToString(tmp["subtype"])
  751. if subtype == "合同" {
  752. wuye["isfirsthand"] = 62
  753. if buyer != "" {
  754. sql := `{
  755. "query": {
  756. "bool": {
  757. "must": [
  758. {
  759. "term": {
  760. "buyer": "` + buyer + `"
  761. }
  762. },
  763. {
  764. "term": {
  765. "tag_topinformation": "情报_物业"
  766. }
  767. },
  768. {
  769. "term": {
  770. "subtype": "合同"
  771. }
  772. },
  773. {
  774. "range": {
  775. "publishtime": {
  776. "lte": ` + fmt.Sprint(publishtime) + `
  777. }
  778. }
  779. }
  780. ]
  781. }
  782. },
  783. "sort": {
  784. "publishtime": "asc"
  785. },
  786. "_source": [
  787. "s_winner"
  788. ],
  789. "size": 10000
  790. }`
  791. data := Es.Get("bidding", "bidding", sql)
  792. if data != nil && len(*data) > 0 {
  793. count := 0
  794. first := util.ObjToString((*data)[0]["s_winner"])
  795. for k, v := range *data {
  796. winner := util.ObjToString(v["s_winner"])
  797. if k > 0 && first != winner {
  798. first = winner
  799. count++
  800. }
  801. }
  802. changehand := fmt.Sprintf("%.2f", float64(count)/float64(len(*data)))
  803. changehands, _ := strconv.ParseFloat(changehand, 64)
  804. wuye["changehand"] = changehands
  805. if changehands > 0.3 {
  806. wuye["changehandindex"] = 61
  807. }
  808. if len(*data) > 1 {
  809. wuye["isfirsthand"] = 0
  810. }
  811. }
  812. }
  813. wuye["period"] = getperiod(compare)
  814. } else if toptype == "招标" || toptype == "采购意向" {
  815. bidamount = util.Float64All(compare["budget"])
  816. }
  817. if tmp["projectinfo"] != nil {
  818. projectInfo := util.ObjToMap(tmp["projectinfo"])
  819. if projectInfo != nil && len(*projectInfo) > 0 {
  820. if (*projectInfo)["attachments"] != nil {
  821. wuye["isfile"] = 63
  822. }
  823. }
  824. }
  825. wuye["scale"] = getBidamountRange(bidamount)
  826. if tmp["property_form"] != nil {
  827. property_form := util.ObjArrToStringArr(tmp["property_form"].([]interface{}))
  828. wuye["property_form"] = getpropertyform(property_form)
  829. }
  830. tagSet["wuye"] = wuye
  831. return tagSet
  832. }
  833. func getBidamountRange(value float64) int {
  834. switch {
  835. case value < 500000:
  836. return 1
  837. case value >= 500000 && value < 1000000:
  838. return 2
  839. case value >= 1000000 && value < 2000000:
  840. return 3
  841. case value >= 2000000 && value < 5000000:
  842. return 4
  843. default:
  844. return 5
  845. }
  846. }
  847. func getpropertyform(value []string) string {
  848. arr := []string{}
  849. categories := map[string]string{
  850. "住宅": "21",
  851. "政府办公楼": "22",
  852. "学校": "23",
  853. "医院": "24",
  854. "产业园区": "25",
  855. "旅游景区": "26",
  856. "交通运输": "27",
  857. "商务办公楼": "28",
  858. "酒店": "29",
  859. }
  860. for _, v := range value {
  861. if categories[v] != "" {
  862. arr = append(arr, categories[v])
  863. }
  864. }
  865. return strings.Join(arr, ",")
  866. }
  867. func getperiod(data map[string]interface{}) int {
  868. res := 16
  869. signaturedate := util.Int64All(data["signaturedate"]) //合同签订日期
  870. expiredate := util.Int64All(data["expiredate"]) //合同截止日期
  871. // contractperiod := util.ObjToString(data["contractperiod"]) //合同期限
  872. project_duration := util.IntAll(data["project_duration"]) //工期时长
  873. project_timeunit := util.ObjToString(data["project_timeunit"]) //工期单位
  874. result := float64(0)
  875. if expiredate > 0 && signaturedate > 0 {
  876. result = calculateYearDifference(signaturedate, expiredate)
  877. } else if project_duration > 0 && project_timeunit != "" {
  878. if strings.Contains(project_timeunit, "年") {
  879. if project_duration == 1 {
  880. res = 12
  881. } else if project_duration == 2 {
  882. res = 13
  883. } else if project_duration == 3 {
  884. res = 14
  885. } else if project_duration == 5 {
  886. res = 15
  887. }
  888. return res
  889. } else if strings.Contains(project_timeunit, "月") {
  890. result = float64(project_duration) / 12
  891. } else if strings.Contains(project_timeunit, "周") {
  892. result = float64(project_duration) * 7 / 365
  893. } else if strings.Contains(project_timeunit, "日") || strings.Contains(project_timeunit, "天") {
  894. result = float64(project_duration) / 365
  895. }
  896. }
  897. if result == 0 {
  898. res = 16
  899. } else if result < 1 {
  900. res = 11
  901. } else if result >= 1 && result < 2 {
  902. res = 12
  903. } else if result >= 2 && result < 3 {
  904. res = 13
  905. } else if result >= 3 && result < 4 {
  906. res = 14
  907. } else if result >= 5 {
  908. res = 15
  909. }
  910. return res
  911. }
  912. func calculateYearDifference(startTime int64, endTime int64) float64 {
  913. start := time.Unix(startTime, 0)
  914. end := time.Unix(endTime, 0)
  915. duration := end.Sub(start)
  916. years := duration.Hours() / 24 / 365
  917. return years
  918. }
  919. func getStr(b string) string {
  920. if b == "" {
  921. return "其它"
  922. }
  923. a1 := "(交通|运输物流|工信|农业|住建|城管|市政|出版广电|检察院|科技|民政|生态环境|市场监管|水利|应急管理|自然资源|财政|档案|党委办|组织|发改|宣传|政府办|政务中心|人大|政协|法院|公安|国资委|海关|机关事务|纪委|军队|人社|商务|审计税务|司法|体育|统计|统战|文旅|民宗|银保监|证监|气象|社会团体|公共资源交易)"
  924. a2 := "(卫健委|医疗)"
  925. a3 := "(教育|学校)"
  926. a4 := "(人行|金融业)"
  927. a5 := "(信息技术|电信行业|农林牧渔|建筑业|传媒|制造业|住宿餐饮|采矿业|能源化工|批发零售)"
  928. if strings.Contains(a1, b) {
  929. return "政府机构"
  930. } else if strings.Contains(a2, b) {
  931. return "医疗单位"
  932. } else if strings.Contains(a3, b) {
  933. return "教育单位"
  934. } else if strings.Contains(a4, b) {
  935. return "金融企业"
  936. } else if strings.Contains(a5, b) {
  937. return "商业公司"
  938. } else {
  939. return "其它"
  940. }
  941. return "其它"
  942. }