task.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746
  1. package main
  2. import (
  3. "encoding/json"
  4. "field_sync/config"
  5. "field_sync/oss"
  6. "fmt"
  7. "go.mongodb.org/mongo-driver/bson"
  8. "go.uber.org/zap"
  9. util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  10. "jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
  11. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  12. "jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
  13. "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
  14. "net"
  15. "reflect"
  16. "regexp"
  17. "sort"
  18. "strings"
  19. "time"
  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. if stype == "bidding" {
  29. uq := bson.M{"gtid": util.ObjToString(mapInfo["gtid"]), "lteid": util.ObjToString(mapInfo["lteid"])}
  30. MgoB.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess": 7, "updatetime": time.Now().Unix()}}, false, true)
  31. }
  32. //领域标签处理的数据 id段
  33. if stype == "bidding_history" {
  34. MgoB.Save("field_data_record", map[string]interface{}{"gtid": mapInfo["gtid"], "lteid": mapInfo["lteid"], "status": 0})
  35. }
  36. q, _ := mapInfo["query"].(map[string]interface{})
  37. bkey, _ := mapInfo["bkey"].(string)
  38. if q == nil {
  39. q = map[string]interface{}{
  40. "_id": map[string]interface{}{
  41. "$gt": mongodb.StringTOBsonId(mapInfo["gtid"].(string)),
  42. "$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
  43. },
  44. }
  45. }
  46. //extract库
  47. extractConn := MgoE.GetMgoConn()
  48. defer MgoE.DestoryMongoConn(extractConn)
  49. extractResult := extractConn.DB(MgoE.DbName).C(config.Conf.DB.MongoE.Coll).Find(q).Select(map[string]interface{}{
  50. "field_source": 0,
  51. "kvtext": 0,
  52. }).Sort("_id").Iter()
  53. eMap := map[string]map[string]interface{}{}
  54. extCount, repeatCount := 0, 0
  55. for tmp := make(map[string]interface{}); extractResult.Next(tmp); extCount++ {
  56. if util.IntAll(tmp["repeat"]) == 1 {
  57. repeatCount++
  58. }
  59. tid := mongodb.BsonIdToSId(tmp["_id"])
  60. eMap[tid] = tmp
  61. tmp = make(map[string]interface{})
  62. }
  63. log.Info("抽取表", zap.Int("数据量", extCount), zap.Int("重复数据量", repeatCount))
  64. //bidding库
  65. biddingConn := MgoB.GetMgoConn()
  66. count, _ := biddingConn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(&q).Count()
  67. log.Info("bidding表", zap.Int64("同步总数:", count))
  68. c := 0
  69. if count < 500000 {
  70. var res []map[string]interface{}
  71. result := biddingConn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(q).Select(map[string]interface{}{
  72. "contenthtml": 0,
  73. }).Iter()
  74. for tmp := make(map[string]interface{}); result.Next(tmp); {
  75. res = append(res, tmp)
  76. tmp = make(map[string]interface{})
  77. }
  78. MgoB.DestoryMongoConn(biddingConn)
  79. log.Info("查询结果", zap.Int64("bidding", count), zap.Int("抽取:", extCount))
  80. c = doIndex(res, eMap, bkey, stype)
  81. } else {
  82. log.Info("查询结果", zap.Int64("数据量太大,放弃", count))
  83. MgoB.DestoryMongoConn(biddingConn)
  84. }
  85. log.Info("bidding sync...over", zap.Int64("all", count), zap.Int("extract sync", c))
  86. NextNode(mapInfo, stype)
  87. NextNodePro(mapInfo, stype)
  88. NextNodeTidb(mapInfo, stype)
  89. if stype == "bidding_history" {
  90. NextNodeBidData(mapInfo) // bidding-data数据
  91. NextNodeTidbQyxy(mapInfo) // tidb-企业数据
  92. NextNodeHn(mapInfo)
  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. update["repeat_id"] = compare["repeat_id"]
  165. } else {
  166. update["extracttype"] = 1
  167. update["dataprocess"] = 8
  168. }
  169. break
  170. } else {
  171. if cid < tid {
  172. compare = nil
  173. continue
  174. } else {
  175. break
  176. }
  177. }
  178. } else {
  179. break
  180. }
  181. }
  182. //------------------对比结束
  183. //处理分类
  184. if compare != nil { //extract
  185. fieldFun(compare, update)
  186. compare = nil
  187. }
  188. // entidlist
  189. extractMap := make(map[string]interface{})
  190. if update["s_winner"] != "" {
  191. cid := companyFun(update)
  192. if len(cid) > 0 {
  193. update["entidlist"] = cid
  194. extractMap["entidlist"] = cid
  195. }
  196. }
  197. if len(extractMap) > 0 {
  198. updateExtPool <- []map[string]interface{}{
  199. {"_id": tmp["_id"]},
  200. {"$set": extractMap},
  201. }
  202. }
  203. // 附件有效字段
  204. if i := validFile(tmp); i != 0 {
  205. if i == -1 {
  206. update["isValidFile"] = false
  207. } else {
  208. update["isValidFile"] = true
  209. }
  210. }
  211. if len(update) > 0 {
  212. if len(del) > 0 { //删除的字段
  213. updateBidPool <- []map[string]interface{}{{
  214. "_id": tmp["_id"],
  215. },
  216. {"$set": update, "$unset": del},
  217. }
  218. } else {
  219. updateBidPool <- []map[string]interface{}{{
  220. "_id": tmp["_id"],
  221. },
  222. {"$set": update},
  223. }
  224. }
  225. }
  226. if count%50000 == 0 {
  227. log.Info("biddingTask", zap.Int("current", count))
  228. }
  229. tmp = make(map[string]interface{})
  230. }
  231. log.Info("biddingAll sync...over", zap.Int("all", count))
  232. }
  233. func doIndex(infos []map[string]interface{}, eMap map[string]map[string]interface{}, bkey, stype string) int {
  234. syncNo := 0 //抽取表数据同步数量
  235. //对比两张表数据,减少查询次数
  236. var compare map[string]interface{}
  237. var bidUpdate [][]map[string]interface{}
  238. var extUpdate [][]map[string]interface{}
  239. //SaveEsLock := &sync.Mutex{}
  240. log.Info("start ...")
  241. for n, tmp := range infos {
  242. tid := mongodb.BsonIdToSId(tmp["_id"])
  243. update := map[string]interface{}{} //要更新的mongo数据
  244. del := map[string]interface{}{}
  245. //对比方法----------------
  246. if eMap[tid] != nil {
  247. compare = eMap[tid]
  248. if stype == "bidding" {
  249. // 增量id段 正常数据
  250. if dg := util.IntAll(compare["dataging"]); dg == 1 { //extract中dataging=1跳过
  251. tmp = make(map[string]interface{})
  252. compare = nil
  253. continue
  254. }
  255. delete(eMap, tid)
  256. }
  257. if stype == "bidding_history" {
  258. //增量id段 历史数据
  259. if compare["history_updatetime"] == nil { //extract中history_updatetime不存在跳过
  260. tmp = make(map[string]interface{})
  261. compare = nil
  262. continue
  263. }
  264. delete(eMap, tid)
  265. }
  266. syncNo++
  267. for _, k := range config.Conf.Serve.FieldS {
  268. v1 := compare[k] //extract
  269. v2 := tmp[k] //bidding
  270. if v2 == nil && v1 != nil {
  271. update[k] = v1
  272. } else if v2 != nil && v1 != nil {
  273. update[k] = v1
  274. } else if v2 != nil && v1 == nil {
  275. if k == "s_subscopeclass" && del["subscopeclass"] == nil {
  276. continue
  277. } else if k == "s_topscopeclass" && del["topscopeclass"] == nil {
  278. continue
  279. } else if k == "city" || k == "district" {
  280. update[k] = ""
  281. } else {
  282. del[k] = 1
  283. }
  284. }
  285. }
  286. // 附件重采,数据同步时不更新判重标识
  287. if util.IntAll(compare["repeat"]) == 1 {
  288. update["extracttype"] = -1
  289. update["dataprocess"] = 7
  290. update["repeat_id"] = compare["repeat_id"]
  291. } else {
  292. update["extracttype"] = 1
  293. update["dataprocess"] = 8
  294. }
  295. } else {
  296. compare = nil
  297. if util.IntAll(tmp["dataging"]) == 1 { //修改未抽取的bidding数据的dataging
  298. update["dataging"] = 0
  299. }
  300. update["dataprocess"] = 8
  301. }
  302. //下面可以多线程跑的--->
  303. //处理分类
  304. if compare != nil { //extract
  305. fieldFun(compare, update)
  306. // publishtime 20230523
  307. if util.IntAll(tmp["publishtime"]) == -1 {
  308. if pb := methodPb(compare); pb > 0 {
  309. update["publishtime"] = pb
  310. }
  311. }
  312. compare = nil
  313. }
  314. //------------------对比结束
  315. //处理key descript
  316. //if bkey == "" {
  317. // DealInfo(&tmp, &update)
  318. //}
  319. // entidlist
  320. extractMap := make(map[string]interface{})
  321. if update["s_winner"] != "" {
  322. cid := companyFun(update)
  323. if len(cid) > 0 {
  324. tmp["entidlist"] = cid
  325. update["entidlist"] = cid
  326. extractMap["entidlist"] = cid
  327. }
  328. }
  329. // 6.10 剑鱼发布信息分类处理, 写在这里是为了修改抽取表
  330. typeFunc(tmp, update, extractMap)
  331. if len(extractMap) > 0 {
  332. if extractMap["toptype"] != nil && extractMap["subtype"] == nil {
  333. extUpdate = append(extUpdate, []map[string]interface{}{
  334. {"_id": tmp["_id"]},
  335. {"$set": extractMap, "$unset": map[string]interface{}{"subtype": ""}},
  336. })
  337. } else {
  338. extUpdate = append(extUpdate, []map[string]interface{}{
  339. {"_id": tmp["_id"]},
  340. {"$set": extractMap},
  341. })
  342. }
  343. if len(extUpdate) >= MgoBulkSize {
  344. tmps := extUpdate
  345. MgoE.UpdateBulk(config.Conf.DB.MongoE.Coll, tmps...)
  346. extUpdate = [][]map[string]interface{}{}
  347. }
  348. }
  349. // 附件有效字段
  350. if i := validFile(tmp); i != 0 {
  351. if i == -1 {
  352. tmp["isValidFile"] = false
  353. update["isValidFile"] = false
  354. } else {
  355. tmp["isValidFile"] = true
  356. update["isValidFile"] = true
  357. }
  358. }
  359. if len(update) > 0 {
  360. if len(del) > 0 {
  361. bidUpdate = append(bidUpdate, []map[string]interface{}{{
  362. "_id": tmp["_id"],
  363. },
  364. {"$set": update, "$unset": del},
  365. })
  366. } else {
  367. bidUpdate = append(bidUpdate, []map[string]interface{}{{
  368. "_id": tmp["_id"],
  369. },
  370. {"$set": update},
  371. })
  372. }
  373. if len(bidUpdate) >= MgoBulkSize {
  374. tmps := bidUpdate
  375. MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, tmps...)
  376. bidUpdate = [][]map[string]interface{}{}
  377. }
  378. }
  379. if n%500 == 0 {
  380. log.Info("biddingTask", zap.Int("current", n))
  381. }
  382. tmp = make(map[string]interface{})
  383. }
  384. //SaveEsLock.Lock()
  385. if len(bidUpdate) > 0 {
  386. tmps := bidUpdate
  387. MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, tmps...)
  388. bidUpdate = [][]map[string]interface{}{}
  389. }
  390. if len(extUpdate) > 0 {
  391. tmps := extUpdate
  392. MgoE.UpdateBulk(config.Conf.DB.MongoE.Coll, tmps...)
  393. extUpdate = [][]map[string]interface{}{}
  394. }
  395. //SaveEsLock.Unlock()
  396. return syncNo
  397. }
  398. // @Description subscopeclass、topscopeclass、package
  399. // 20230523 多包处理 subpackage = 1
  400. // @Author J 2022/6/7 5:54 PM
  401. func fieldFun(compare, update map[string]interface{}) {
  402. subscopeclass, _ := compare["subscopeclass"].([]interface{}) //subscopeclass
  403. if subscopeclass != nil {
  404. m1 := map[string]bool{}
  405. newclass := []string{}
  406. for _, sc := range subscopeclass {
  407. sclass, _ := sc.(string)
  408. if !m1[sclass] {
  409. m1[sclass] = true
  410. newclass = append(newclass, sclass)
  411. }
  412. }
  413. update["s_subscopeclass"] = strings.Join(newclass, ",")
  414. update["subscopeclass"] = newclass
  415. }
  416. topscopeclass, _ := compare["topscopeclass"].([]interface{}) //topscopeclass
  417. if topscopeclass != nil {
  418. m2 := map[string]bool{}
  419. newclass := []string{}
  420. for _, tc := range topscopeclass {
  421. tclass, _ := tc.(string)
  422. tclass = regLetter.ReplaceAllString(tclass, "") // 去除字母
  423. if !m2[tclass] {
  424. m2[tclass] = true
  425. newclass = append(newclass, tclass)
  426. }
  427. }
  428. update["topscopeclass"] = topscopeclass
  429. update["s_topscopeclass"] = strings.Join(newclass, ",")
  430. }
  431. if package1 := compare["package"]; package1 != nil {
  432. packageM, _ := package1.(map[string]interface{})
  433. update["package"] = packageM
  434. for _, p := range packageM {
  435. pm, _ := p.(map[string]interface{})
  436. if util.ObjToString(pm["winner"]) != "" || util.Float64All(pm["budget"]) > 0 ||
  437. util.Float64All(pm["bidamount"]) > 0 {
  438. update["multipackage"] = 1
  439. break
  440. }
  441. }
  442. } else {
  443. update["multipackage"] = 0
  444. }
  445. // subpackage
  446. if compare["package"] != nil && compare["s_winner"] != nil && compare["bidamount"] != nil {
  447. pg := compare["package"].(map[string]interface{})
  448. if len(pg) > 1 {
  449. var bmt []float64
  450. var swn []string
  451. for _, p := range pg {
  452. p1 := p.(map[string]interface{})
  453. if p1["bidamount"] != nil {
  454. bmt = append(bmt, util.Float64All(p1["bidamount"]))
  455. }
  456. if w := util.ObjToString(p1["winner"]); w != "" {
  457. swn = append(swn)
  458. }
  459. }
  460. if len(bmt) > 1 && len(swn) > 1 {
  461. sn := strings.Split(util.ObjToString(compare["s_winner"]), ",")
  462. sort.Strings(sn)
  463. sort.Strings(swn)
  464. swn1 := util.ObjArrToStringArr(Duplicate(swn)) // 去重
  465. if strings.Join(swn1, ",") == strings.Join(sn, ",") {
  466. bidamount := 0.0
  467. for _, f := range bmt {
  468. bidamount += f
  469. }
  470. if bidamount == util.Float64All(compare["bidamount"]) {
  471. update["subpackage"] = 1
  472. }
  473. }
  474. }
  475. }
  476. }
  477. }
  478. // @Description entidlist
  479. // @Author J 2022/6/7 2:36 PM
  480. func companyFun(tmp map[string]interface{}) (cid []string) {
  481. sWinnerarr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
  482. for _, w := range sWinnerarr {
  483. if w != "" {
  484. id := redis.GetStr("qyxy_id", w)
  485. if id == "" {
  486. 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)
  487. if len(*ents) > 0 {
  488. id = util.ObjToString((*ents)[0]["_id"])
  489. redis.PutCKV("qyxy_id", w, id)
  490. } else {
  491. ent, _ := MgoP.FindOne(config.Conf.DB.MongoP.Coll, map[string]interface{}{"history_name": w})
  492. if len(*ent) > 0 {
  493. id = util.ObjToString((*ent)["company_id"])
  494. redis.PutCKV("qyxy_id", w, id)
  495. }
  496. }
  497. }
  498. if id == "" {
  499. id = "-"
  500. }
  501. cid = append(cid, id)
  502. }
  503. }
  504. return cid
  505. }
  506. // @Description update 修改bidding表,extractM修改抽取表
  507. // @Author J 2022/6/10 10:29 AM
  508. func typeFunc(tmp, update, extractM map[string]interface{}) {
  509. if jyData, ok := tmp["jyfb_data"].(map[string]interface{}); ok {
  510. if t := util.ObjToString(jyData["type"]); t != "" {
  511. switch t {
  512. //case "采购信息":
  513. case "招标公告":
  514. if util.ObjToString(tmp["toptype"]) != "招标" {
  515. update["toptype"] = "招标"
  516. extractM["toptype"] = "招标"
  517. delete(update, "subtype")
  518. }
  519. case "采购意向":
  520. if util.ObjToString(tmp["toptype"]) != "采购意向" {
  521. update["toptype"] = "采购意向"
  522. update["subtype"] = "采购意向"
  523. extractM["toptype"] = "采购意向"
  524. extractM["subtype"] = "采购意向"
  525. }
  526. case "招标预告":
  527. if util.ObjToString(tmp["toptype"]) != "预告" {
  528. update["toptype"] = "预告"
  529. extractM["toptype"] = "预告"
  530. delete(update, "subtype")
  531. }
  532. case "招标结果":
  533. if util.ObjToString(tmp["toptype"]) != "结果" {
  534. update["toptype"] = "结果"
  535. extractM["toptype"] = "结果"
  536. delete(update, "subtype")
  537. }
  538. }
  539. }
  540. }
  541. }
  542. // @Description 附件有效字段(isValidFile)
  543. // @Author J 2022/7/8 14:41
  544. func validFile(tmp map[string]interface{}) int {
  545. isContinue := false
  546. if pinfo, o := tmp["projectinfo"].(map[string]interface{}); o {
  547. if atts, o1 := pinfo["attachments"].(map[string]interface{}); o1 {
  548. for _, att := range atts {
  549. if att == nil {
  550. continue
  551. }
  552. if reflect.TypeOf(att).String() == "string" {
  553. continue
  554. }
  555. att1 := att.(map[string]interface{})
  556. if fid := util.ObjToString(att1["fid"]); fid != "" {
  557. isContinue = true
  558. break
  559. }
  560. }
  561. if isContinue {
  562. if attachTxt, o := tmp["attach_text"].(map[string]interface{}); o {
  563. if len(attachTxt) > 0 {
  564. for _, at := range attachTxt {
  565. at1 := at.(map[string]interface{})
  566. if len(at1) > 0 {
  567. for k, _ := range at1 {
  568. if reflect.TypeOf(at1[k]).String() == "string" {
  569. continue
  570. }
  571. at2 := at1[k].(map[string]interface{})
  572. s := strings.ToLower(util.ObjToString(at2["file_name"]))
  573. if !strings.Contains(s, "jpg") || !strings.Contains(s, "jpeg") != strings.Contains(s, "png") ||
  574. strings.Contains(s, "pdf") {
  575. if strings.Contains(s, "swf") || strings.Contains(s, "html") {
  576. return -1
  577. } else if AnalysisFile(oss.OssGetObject(util.ObjToString(at2["attach_url"]))) {
  578. return 1
  579. }
  580. }
  581. }
  582. break
  583. } else {
  584. break
  585. }
  586. }
  587. }
  588. }
  589. flag := false
  590. for _, att := range atts {
  591. if att == nil {
  592. continue
  593. }
  594. if reflect.TypeOf(att).String() == "string" {
  595. continue
  596. }
  597. att1 := att.(map[string]interface{})
  598. if fid := util.ObjToString(att1["fid"]); fid != "" {
  599. ftype := strings.ToLower(util.ObjToString(tmp["ftype"]))
  600. if ftype != "swf" && ftype != "html" && oss.OssObjExists("jy-datafile", fid) {
  601. return 1
  602. } else {
  603. flag = true
  604. }
  605. }
  606. }
  607. if flag {
  608. return -1
  609. }
  610. }
  611. }
  612. }
  613. return 0
  614. }
  615. // @Description id不变,内容变化 重新索引数据
  616. // @Author J 2022/8/10 13:29
  617. func taskinfo(id string) {
  618. tmp, _ := MgoB.FindById("bidding", id, nil)
  619. if tmp == nil || len(*tmp) == 0 {
  620. log.Info(fmt.Sprintf("taskinfo bidding id=%s 未查询到数据", id))
  621. return
  622. }
  623. extractM, _ := MgoE.FindById(config.Conf.DB.MongoE.Coll, id, nil)
  624. if extractM == nil || len(*extractM) == 0 {
  625. extractM, _ = MgoE.FindById(config.Conf.DB.MongoE.Coll1, id, nil)
  626. if extractM == nil || len(*extractM) == 0 {
  627. log.Info(fmt.Sprintf("taskinfo extract id=%s 未查询到数据", id))
  628. return
  629. }
  630. }
  631. update := map[string]interface{}{} //要更新的mongo数据
  632. //更新bidding表字段
  633. for _, k := range config.Conf.Serve.FieldS {
  634. v1 := (*extractM)[k] //extract
  635. v2 := (*tmp)[k] //bidding
  636. if v2 == nil && v1 != nil {
  637. update[k] = v1
  638. } else if v2 != nil && v1 != nil {
  639. update[k] = v1
  640. } else if v2 != nil && v1 == nil {
  641. if k == "city" || k == "district" {
  642. update[k] = ""
  643. }
  644. }
  645. }
  646. if util.IntAll((*extractM)["repeat"]) == 1 {
  647. update["extracttype"] = -1
  648. update["dataprocess"] = 7
  649. update["repeat_id"] = (*extractM)["repeat_id"]
  650. } else {
  651. update["extracttype"] = 1
  652. update["dataprocess"] = 8
  653. }
  654. //处理分类
  655. fieldFun(*extractM, update)
  656. extractMap := make(map[string]interface{})
  657. if util.ObjToString((*tmp)["s_winner"]) != "" {
  658. cid := companyFun(*tmp)
  659. if len(cid) > 0 {
  660. update["entidlist"] = cid
  661. extractMap["entidlist"] = cid
  662. }
  663. MgoE.UpdateById(config.Conf.DB.MongoE.Coll, id, map[string]interface{}{"$set": extractMap})
  664. }
  665. // 附件有效字段
  666. if i := validFile(*tmp); i != 0 {
  667. if i == -1 {
  668. update["isValidFile"] = false
  669. } else {
  670. update["isValidFile"] = true
  671. }
  672. }
  673. if len(update) > 0 {
  674. MgoB.UpdateById(config.Conf.DB.MongoB.Coll, id, map[string]interface{}{"$set": update})
  675. }
  676. mapinfo := map[string]interface{}{
  677. "infoid": id,
  678. "stype": "index-by-id",
  679. }
  680. datas, _ := json.Marshal(mapinfo)
  681. var next = &net.UDPAddr{
  682. IP: net.ParseIP(config.Conf.Udp.Next.Addr),
  683. Port: util.IntAll(config.Conf.Udp.Next.Port),
  684. }
  685. log.Info("nsq data over", zap.Any("es", next), zap.String("mapinfo", string(datas)))
  686. _ = UdpClient.WriteUdp(datas, udp.OP_TYPE_DATA, next)
  687. }
  688. var DateTimeSelect = []string{"bidopentime", "bidendtime", "signaturedate", "comeintime"}
  689. // @Description 发布时间处理
  690. // @Author J 2023/5/23 14:32
  691. func methodPb(tmp map[string]interface{}) int64 {
  692. if tmp["ext_publishtime"] != nil {
  693. if newPb := util.Int64All(tmp["ext_publishtime"]); newPb < time.Now().Unix() && newPb > 1420041600 {
  694. return newPb
  695. }
  696. }
  697. for _, d := range DateTimeSelect {
  698. if tmp[d] != nil && util.Int64All(tmp[d]) < time.Now().Unix() {
  699. return util.Int64All(tmp[d])
  700. }
  701. }
  702. return 0
  703. }
  704. // Duplicate
  705. // @Description 去重
  706. // @Author J 2023/5/24 09:53
  707. func Duplicate(a interface{}) (ret []interface{}) {
  708. va := reflect.ValueOf(a)
  709. for i := 0; i < va.Len(); i++ {
  710. if i > 0 && reflect.DeepEqual(va.Index(i-1).Interface(), va.Index(i).Interface()) {
  711. continue
  712. }
  713. ret = append(ret, va.Index(i).Interface())
  714. }
  715. return ret
  716. }