task.go 28 KB

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