task.go 26 KB

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