timedTaskBuyer.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665
  1. package main
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "github.com/garyburd/redigo/redis"
  6. "go.mongodb.org/mongo-driver/bson/primitive"
  7. "gopkg.in/mgo.v2/bson"
  8. "log"
  9. mu "mfw/util"
  10. "net"
  11. "qfw/util"
  12. "sort"
  13. "strings"
  14. "time"
  15. )
  16. //之前main方法,只更新
  17. func TaskBuyer(mapinfo *map[string]interface{}) {
  18. defer util.Catch()
  19. gtid, lteid := util.ObjToString((*mapinfo)["gtid"]), util.ObjToString((*mapinfo)["lteid"])
  20. if gtid == "" || lteid == "" {
  21. log.Println(gtid, lteid, "参数错误")
  22. return
  23. }
  24. GId, err := primitive.ObjectIDFromHex(gtid)
  25. LtId, err2 := primitive.ObjectIDFromHex(lteid)
  26. if err != nil || err2 != nil {
  27. log.Println(gtid, lteid, "转换_id错误")
  28. return
  29. }
  30. //udp的id区间查询bidding 中标人 中标联系人 中标联系电话
  31. // topscopeclass项目类型-industry行业类型&&topscopeclass联系人项目类型
  32. // (area地区-province省份 city城市-city城市 district区县-district区县)
  33. // winneraddr-company_address企业地址
  34. SourceClientcc := SourceClient.GetMgoConn()
  35. defer SourceClient.DestoryMongoConn(SourceClientcc)
  36. cursor := SourceClientcc.DB(Config["mgodb_bidding"]).C(Config["mgodb_mgoinit_c"]).Find(bson.M{
  37. "_id": bson.M{
  38. "$gte": GId,
  39. "$lte": LtId,
  40. },
  41. }).Select(bson.M{"buyer": 1, "buyertel": 1, "buyerperson": 1,
  42. "topscopeclass": 1, "buyeraddr": 1}).Iter()
  43. if cursor == nil {
  44. log.Println(cursor)
  45. return
  46. }
  47. //判断是否是存量,是存量走Redis遍历
  48. if v, ok := (*mapinfo)["data_info"].(string); ok && v == "save" {
  49. //存量处理
  50. tmp := map[string]interface{}{}
  51. conn := HisRedisPool.Conn()
  52. defer conn.Close()
  53. //选择redis db
  54. conn.Select(1)
  55. //遍历bidding表保存到redis
  56. // key:_id value:json结构体
  57. for cursor.Next(&tmp) {
  58. if tmp["buyer"] == nil || tmp["buyer"] == "" {
  59. continue
  60. }
  61. mgoId:=tmp["_id"].(primitive.ObjectID).Hex()
  62. delete(tmp,"_id")
  63. bytes, _ := json.Marshal(tmp)
  64. if err := conn.Set(mgoId, string(bytes), 0).Err(); err != nil {
  65. log.Println(err)
  66. }
  67. }
  68. //遍历redis
  69. if scan := conn.Scan(0, "", 100); scan.Err() != nil {
  70. log.Println(scan.Err())
  71. return
  72. } else {
  73. iterator := scan.Iterator()
  74. for iterator.Next() {
  75. redisId := iterator.Val() //redis key
  76. redisvalue := conn.Get(iterator.Val()).Val() //redis val
  77. tmp := make(map[string]interface{})
  78. json.Unmarshal([]byte(redisvalue),&tmp)
  79. //重复增量操作
  80. //redis查询是否存在
  81. rdb := RedisPool.Get()
  82. rdb.Do("SELECT","1")
  83. if reply, err := redis.String(rdb.Do("GET", tmp["buyer"])); err != nil {
  84. //redis不存在,存到临时表,定时任务处理
  85. FClient.DbName = Config["mgodb_extract_kf"]
  86. if tmpid := FClient.Save("buyer_new", tmp); tmpid == nil {
  87. log.Println("存量 FClient.Save err", tmpid)
  88. }
  89. //log.Println("get redis id err:定时任务处理", err, tmp)
  90. if err := rdb.Close(); err != nil {
  91. log.Println("存量",err)
  92. }
  93. //删除存量redis
  94. conn.Del(redisId)
  95. continue
  96. } else {
  97. if err := rdb.Close(); err != nil {
  98. log.Println(err)
  99. }
  100. //拿到合并后的qyk
  101. FClient.DbName = Config["mgodb_extract_kf"]
  102. oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
  103. if oldTmp == nil {
  104. log.Println("存量 redis id 不存在",reply,tmp["buyer"])
  105. continue
  106. }
  107. tmpTopscopeclass := []string{}
  108. tmpTopscopeclassMap := make(map[string]bool)
  109. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  110. for _, vv := range v {
  111. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  112. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  113. }
  114. }
  115. for k := range tmpTopscopeclassMap {
  116. tmpTopscopeclass = append(tmpTopscopeclass, k)
  117. }
  118. }
  119. sort.Strings(tmpTopscopeclass)
  120. //更新buyerclass
  121. esId := oldTmp["_id"].(primitive.ObjectID).Hex()
  122. //更新行业类型
  123. if tmp["buyerperson"] == nil || tmp["buyerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["buyerperson"])) {
  124. //更新buyerclass合并
  125. if tmp["buyerclass"] == nil || tmp["buyerclass"] == "" {
  126. //无值,不更新
  127. }else {
  128. //有值
  129. var buyerclass_new,buyerclass_old string
  130. buyerclass_new = tmp["buyerclass"].(string)
  131. buyerclass_old = oldTmp["buyerclass"].(string)
  132. if strings.Contains(buyerclass_old, buyerclass_new) {
  133. oldTmp["buyerclass"] = buyerclass_old + ","+buyerclass_new //采购单位类型
  134. }
  135. }
  136. oldTmp["updatatime"] = time.Now().Unix()
  137. //mongo更新
  138. FClient.DbName = Config["mgodb_extract_kf"]
  139. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  140. log.Println("mongo更新err", esId)
  141. }
  142. //es更新
  143. delete(oldTmp, "_id")
  144. if _, err := EsConn.Update().Index(Config["elasticsearch_buyer_index"]).Type(Config["elasticsearch_buyer_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  145. log.Println("update es err:", err)
  146. }
  147. //删除存量redis
  148. conn.Del(redisId)
  149. continue
  150. }
  151. //联系方式合并
  152. var tmpperson, buyertel string
  153. tmpperson = tmp["buyerperson"].(string)
  154. if tmp["buyertel"] == nil || tmp["buyertel"] == "" {
  155. buyertel = ""
  156. } else {
  157. if Reg_xing.MatchString(util.ObjToString(tmp["buyertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["buyertel"])) {
  158. buyertel = ""
  159. } else {
  160. buyertel = util.ObjToString(tmp["buyertel"])
  161. }
  162. }
  163. contactMaps := make([]interface{}, 0)
  164. if oldTmp["contact"] == nil {
  165. tmpContact := make(map[string]interface{})
  166. tmpContact["infoid"] = redisId
  167. tmpContact["contact_person"] = tmpperson
  168. tmpContact["contact_type"] = "项目联系人"
  169. tmpContact["phone"] = buyertel
  170. tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  171. tmpContact["updatetime"] = time.Now().Unix()
  172. contactMaps = append(contactMaps, tmpContact)
  173. } else {
  174. //对比前四项,相等丢弃
  175. if v, ok := oldTmp["contact"].(primitive.A); ok {
  176. var isNotUpdate bool
  177. for _, vv := range v {
  178. if vvv, ok := vv.(map[string]interface{}); ok {
  179. if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
  180. vvv["phone"] == buyertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
  181. isNotUpdate = true
  182. vvv["updatetime"] = time.Now().Unix()
  183. }
  184. contactMaps = append(contactMaps, vvv)
  185. }
  186. }
  187. if !isNotUpdate {
  188. vvv := make(map[string]interface{})
  189. vvv["infoid"] = redisId
  190. vvv["contact_person"] = tmp["buyerperson"]
  191. vvv["contact_type"] = "项目联系人"
  192. vvv["phone"] = buyertel
  193. vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  194. vvv["updatetime"] = time.Now().Unix()
  195. contactMaps = append(contactMaps, vvv)
  196. }
  197. }
  198. }
  199. oldTmp["contact"] = contactMaps
  200. //更新buyerclass合并
  201. if tmp["buyerclass"] == nil || tmp["buyerclass"] == "" {
  202. //无值,不更新
  203. }else {
  204. //有值
  205. var buyerclass_new,buyerclass_old string
  206. buyerclass_new = tmp["buyerclass"].(string)
  207. buyerclass_old = oldTmp["buyerclass"].(string)
  208. if strings.Contains(buyerclass_old, buyerclass_new) {
  209. oldTmp["buyerclass"] = buyerclass_old + ","+buyerclass_new //采购单位类型
  210. }
  211. }
  212. //mongo更新
  213. oldTmp["updatatime"] = time.Now().Unix()
  214. FClient.DbName = Config["mgodb_extract_kf"]
  215. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  216. log.Println("存量 mongo更新 err", esId, oldTmp)
  217. }
  218. //es更新
  219. delete(oldTmp, "_id")
  220. if _, err := EsConn.Update().Index(Config["elasticsearch_buyer_index"]).Type(Config["elasticsearch_buyer_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  221. log.Println("存量 EsConn err :", err)
  222. }
  223. //最后删除redis
  224. conn.Del(redisId)
  225. }
  226. }
  227. }
  228. log.Println("存量历史合并执行完成 ok", gtid, lteid)
  229. } else {
  230. //增量处理
  231. overid := gtid
  232. tmp := map[string]interface{}{}
  233. for cursor.Next(&tmp) {
  234. overid = tmp["_id"].(primitive.ObjectID).Hex()
  235. log.Println(tmp["_id"])
  236. if tmp["buyer"] == nil || tmp["buyer"] == "" {
  237. continue
  238. }
  239. //redis查询是否存在
  240. rdb := RedisPool.Get()
  241. rdb.Do("SELECT","1")
  242. if reply, err := redis.String(rdb.Do("GET", tmp["buyer"])); err != nil {
  243. //redis不存在存到临时表,定时任务处理
  244. FClient.DbName = Config["mgodb_extract_kf"]
  245. if tmpid := FClient.Save("buyer_new", tmp); tmpid == nil {
  246. log.Println("FClient.Save err", tmpid)
  247. }
  248. //log.Println("get redis id err:定时任务处理", err, tmp)
  249. if err := rdb.Close(); err != nil {
  250. log.Println(err)
  251. }
  252. continue
  253. } else {
  254. if err := rdb.Close(); err != nil {
  255. log.Println(err)
  256. }
  257. //拿到合并后的qyk
  258. FClient.DbName = Config["mgodb_extract_kf"]
  259. oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
  260. if oldTmp == nil {
  261. log.Println("redis id 不存在")
  262. continue
  263. }
  264. //比较合并
  265. //行业类型
  266. tmpTopscopeclass := []string{}
  267. tmpTopscopeclassMap := make(map[string]bool)
  268. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  269. for _, vv := range v {
  270. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  271. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  272. }
  273. }
  274. for k := range tmpTopscopeclassMap {
  275. tmpTopscopeclass = append(tmpTopscopeclass, k)
  276. }
  277. }
  278. sort.Strings(tmpTopscopeclass)
  279. esId := oldTmp["_id"].(primitive.ObjectID).Hex()
  280. //更新行业类型 buyerclass合并
  281. if tmp["buyerperson"] == nil || tmp["buyerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["buyerperson"])) {
  282. //更新buyerclass合并
  283. if tmp["buyerclass"] == nil || tmp["buyerclass"] == "" {
  284. //无值,不更新
  285. }else {
  286. //有值
  287. var buyerclass_new,buyerclass_old string
  288. buyerclass_new = tmp["buyerclass"].(string)
  289. buyerclass_old = oldTmp["buyerclass"].(string)
  290. if strings.Contains(buyerclass_old, buyerclass_new) {
  291. oldTmp["buyerclass"] = buyerclass_old + ","+buyerclass_new //采购单位类型
  292. }
  293. }
  294. oldTmp["updatatime"] = time.Now().Unix()
  295. //mongo更新
  296. FClient.DbName = Config["mgodb_extract_kf"]
  297. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  298. log.Println("mongo更新err", esId)
  299. }
  300. //es更新
  301. delete(oldTmp, "_id")
  302. if _, err := EsConn.Update().Index(Config["elasticsearch_buyer_index"]).Type(Config["elasticsearch_buyer_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  303. log.Println("update es err:", err)
  304. }
  305. continue
  306. }
  307. //联系方式合并
  308. var tmpperson, buyertel string
  309. tmpperson = tmp["buyerperson"].(string)
  310. if tmp["buyertel"] == nil || tmp["buyertel"] == "" {
  311. buyertel = ""
  312. } else {
  313. if Reg_xing.MatchString(util.ObjToString(tmp["buyertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["buyertel"])) {
  314. buyertel = ""
  315. } else {
  316. buyertel = util.ObjToString(tmp["buyertel"])
  317. }
  318. }
  319. contactMaps := make([]interface{}, 0)
  320. if oldTmp["contact"] == nil {
  321. tmpContact := make(map[string]interface{})
  322. tmpContact["infoid"] = overid
  323. tmpContact["contact_person"] = tmpperson
  324. tmpContact["contact_type"] = "项目联系人"
  325. tmpContact["phone"] = buyertel
  326. tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  327. tmpContact["updatetime"] = time.Now().Unix()
  328. contactMaps = append(contactMaps, tmpContact)
  329. } else {
  330. //对比前四项,相等丢弃
  331. if v, ok := oldTmp["contact"].(primitive.A); ok {
  332. var isNotUpdate bool
  333. for _, vv := range v {
  334. if vvv, ok := vv.(map[string]interface{}); ok {
  335. if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
  336. vvv["phone"] == buyertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
  337. isNotUpdate = true
  338. vvv["updatetime"] = time.Now().Unix()
  339. }
  340. contactMaps = append(contactMaps, vvv)
  341. }
  342. }
  343. if !isNotUpdate {
  344. vvv := make(map[string]interface{})
  345. vvv["infoid"] = overid
  346. vvv["contact_person"] = tmp["buyerperson"]
  347. vvv["contact_type"] = "项目联系人"
  348. vvv["phone"] = buyertel
  349. vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  350. vvv["updatetime"] = time.Now().Unix()
  351. contactMaps = append(contactMaps, vvv)
  352. }
  353. }
  354. }
  355. oldTmp["contact"] = contactMaps
  356. //更新buyerclass合并
  357. if tmp["buyerclass"] == nil || tmp["buyerclass"] == "" {
  358. //无值,不更新
  359. }else {
  360. //有值
  361. var buyerclass_new,buyerclass_old string
  362. buyerclass_new = tmp["buyerclass"].(string)
  363. buyerclass_old = oldTmp["buyerclass"].(string)
  364. if strings.Contains(buyerclass_old, buyerclass_new) {
  365. oldTmp["buyerclass"] = buyerclass_old + ","+buyerclass_new //采购单位类型
  366. }
  367. }
  368. //mongo更新
  369. oldTmp["updatatime"] = time.Now().Unix()
  370. FClient.DbName = Config["mgodb_extract_kf"]
  371. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  372. log.Println("mongo更新 err", esId, oldTmp)
  373. }
  374. //es更新
  375. delete(oldTmp, "_id")
  376. if _, err := EsConn.Update().Index(Config["elasticsearch_buyer_index"]).Type(Config["elasticsearch_buyer_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  377. log.Println("EsConn err :", err)
  378. }
  379. }
  380. }
  381. log.Println("增量合并执行完成 ok", gtid, lteid, overid)
  382. }
  383. }
  384. //定时任务
  385. //1.存异常表
  386. //2.合并原始库新增
  387. func TimedTaskBuyer() {
  388. //time.Sleep(time.Hour*70)
  389. t2 := time.NewTimer(time.Second * 5)
  390. for range t2.C {
  391. Fcconn := FClient.GetMgoConn()
  392. defer FClient.DestoryMongoConn(Fcconn)
  393. tmpLast := map[string]interface{}{}
  394. if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("buyer_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
  395. if !iter.Next(&tmpLast) {
  396. //临时表无数据
  397. log.Println("临时表无数据:")
  398. t2.Reset(time.Minute * 5)
  399. continue
  400. } else {
  401. log.Println("临时表有数据:", tmpLast)
  402. fconn := FClient.GetMgoConn()
  403. defer FClient.DestoryMongoConn(fconn)
  404. cursor := fconn.DB(Config["mgodb_extract_kf"]).C("buyer_new").Find(bson.M{
  405. "_id": bson.M{
  406. "$lte": tmpLast["_id"],
  407. },
  408. }).Sort("_id").Iter()
  409. if cursor == nil {
  410. log.Println("查询失败")
  411. t2.Reset(time.Second * 5)
  412. continue
  413. }
  414. //遍历临时表数据,匹配不到原始库存入异常表
  415. tmp := make(map[string]interface{})
  416. for cursor.Next(&tmp) {
  417. tmpId := tmp["_id"].(primitive.ObjectID).Hex()
  418. //再重新查找redis,存在发udp处理,不存在走新增合并
  419. rdb := RedisPool.Get()
  420. rdb.Do("SELECT","1")
  421. if _, err := redis.String(rdb.Do("GET", tmp["buyer"])); err == nil {
  422. //{"gtid":"57d7ad2f61a0721f152d2ad5","lteid":"5e20968d85a9271abf0ad6c2","stype":""}
  423. //redis存在发送udp进行处理
  424. by, _ := json.Marshal(map[string]interface{}{
  425. "gtid": tmpId,
  426. "lteid": tmpId,
  427. "stype": "",
  428. })
  429. if e := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
  430. IP: net.ParseIP("127.0.0.1"),
  431. Port: Updport,
  432. }); e != nil {
  433. log.Println(e)
  434. }
  435. //存在的话删除tmp mongo表
  436. FClient.DbName = Config["mgodb_extract_kf"]
  437. if DeletedCount := FClient.DeleteById("buyer_new", tmpId); DeletedCount == 0 {
  438. log.Println("删除临时表err:", DeletedCount)
  439. }
  440. if err := rdb.Close(); err != nil {
  441. log.Println(err)
  442. }
  443. continue
  444. } else {
  445. if err = rdb.Close(); err != nil {
  446. log.Println(err)
  447. }
  448. }
  449. //查询redis不存在新增
  450. FClient.DbName = Config["mgodb_enterprise"]
  451. //qyxy 企业库 两亿条
  452. resulttmp := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": tmp["buyer"]})
  453. if resulttmp["_id"] == nil {
  454. //log.Println(r)
  455. //匹配不到原始库,存入异常表删除临时表
  456. FClient.DbName = Config["mgodb_extract_kf"]
  457. if saveid := FClient.Save("buyer_err", tmp); saveid == nil {
  458. log.Println("存入异常表错误", tmp)
  459. }
  460. FClient.DbName = Config["mgodb_extract_kf"]
  461. if deleteNum := FClient.DeleteById("buyer_new", tmpId); deleteNum == 0 {
  462. log.Println("删除临时表错误", deleteNum)
  463. }
  464. continue
  465. } else {
  466. //log.Println(123)
  467. //匹配到原始库,新增 resulttmp
  468. if resulttmp["credit_no"] != nil {
  469. if credit_no, ok := resulttmp["credit_no"].(string); ok && strings.TrimSpace(credit_no) != "" &&
  470. len(strings.TrimSpace(credit_no)) > 8 {
  471. dataNo := strings.TrimSpace(credit_no)[2:8]
  472. if Addrs[dataNo] != nil {
  473. if v, ok := Addrs[dataNo].(map[string]interface{}); ok {
  474. if resulttmp["province"] == nil || resulttmp["province"] == "" {
  475. resulttmp["province"] = v["province"]
  476. }
  477. resulttmp["city"] = v["city"]
  478. resulttmp["district"] = v["district"]
  479. }
  480. }
  481. }
  482. }
  483. contacts := make([]map[string]interface{}, 0)
  484. contact := make(map[string]interface{}, 0)
  485. if resulttmp["legal_person"] != nil {
  486. contact["contact_person"] = resulttmp["legal_person"] //联系人
  487. } else {
  488. contact["contact_person"] = "" //联系人
  489. }
  490. contact["contact_type"] = "法定代表人" //法定代表人
  491. //log.Println(1)
  492. if resulttmp["annual_reports"] != nil {
  493. bytes, err := json.Marshal(resulttmp["annual_reports"])
  494. if err != nil {
  495. log.Println("annual_reports err:", err)
  496. }
  497. phonetmp := make([]map[string]interface{}, 0)
  498. err = json.Unmarshal(bytes, &phonetmp)
  499. if err != nil {
  500. log.Println("Unmarshal err:", err)
  501. }
  502. for _, vv := range phonetmp {
  503. if vv["company_phone"] != nil {
  504. if vv["company_phone"] == "" {
  505. continue
  506. } else {
  507. contact["phone"] = vv["company_phone"] //联系电话
  508. break
  509. }
  510. } else {
  511. contact["phone"] = "" //联系电话
  512. }
  513. }
  514. }
  515. //log.Println(k, contact["phone"], resulttmp["_id"])
  516. //time.Sleep(10 * time.Second)
  517. if contact["phone"] == nil {
  518. contact["phone"] = "" //联系电话
  519. }
  520. contact["topscopeclass"] = "企业公示" //项目类型
  521. contact["updatetime"] = time.Now().Unix() //更新时间
  522. contact["infoid"] = "" //招标信息id
  523. contacts = append(contacts, contact)
  524. resulttmp["contact"] = contacts
  525. savetmp := make(map[string]interface{}, 0)
  526. for _, sk := range BuyerFields {
  527. if sk == "_id" {
  528. savetmp["tmp"+sk] = resulttmp[sk]
  529. continue
  530. } else if sk == "area_code" {
  531. //行政区划代码
  532. savetmp[sk] = fmt.Sprint(resulttmp[sk])
  533. continue
  534. } else if sk == "report_websites" {
  535. //网址
  536. if resulttmp["report_websites"] == nil {
  537. savetmp["website"] = ""
  538. } else {
  539. report_websitesArr := []string{}
  540. if ppms, ok := resulttmp[sk].(primitive.A); ok {
  541. for _, v := range ppms {
  542. if vvv, ok := v.(map[string]interface{}); ok {
  543. if rv, ok := vvv["website_url"].(string); ok {
  544. report_websitesArr = append(report_websitesArr, rv)
  545. }
  546. }
  547. }
  548. }
  549. sort.Strings(report_websitesArr)
  550. savetmp["website"] = strings.Join(report_websitesArr, ";")
  551. }
  552. continue
  553. } else if sk == "wechat_accounts" {
  554. savetmp[sk] = []interface{}{}
  555. continue
  556. }else if sk=="buyer_name" {
  557. if resulttmp["company_name"] == nil {
  558. savetmp[sk] = ""
  559. }else {
  560. savetmp[sk] = resulttmp[sk]
  561. }
  562. continue
  563. }else if sk=="address"{
  564. if resulttmp["company_address"] == nil {
  565. savetmp[sk] = ""
  566. }else {
  567. savetmp[sk] = resulttmp[sk]
  568. }
  569. continue
  570. }
  571. if resulttmp[sk] == nil && sk != "history_name" && sk != "wechat_accounts" &&
  572. sk != "buyer_name" && sk != "address" &&
  573. sk != "contact" && sk != "report_websites" {
  574. savetmp[sk] = ""
  575. } else {
  576. savetmp[sk] = resulttmp[sk]
  577. }
  578. }
  579. //tmps = append(tmps, savetmp)
  580. savetmp["updatatime"] = time.Now().Unix()
  581. //保存mongo
  582. FClient.DbName = Config["mgodb_extract_kf"]
  583. saveid := FClient.Save(Config["mgo_qyk_c"], savetmp)
  584. if saveid != nil {
  585. //保存redis
  586. rc := RedisPool.Get()
  587. rc.Do("SELECT","1")
  588. var _id string
  589. if v, ok := saveid.(primitive.ObjectID); ok {
  590. _id = v.Hex()
  591. }
  592. if _, err := rc.Do("SET", savetmp["buyer_name"], _id); err != nil {
  593. log.Println("save redis err:", tmp["_id"], savetmp["_id"], savetmp["buyer_name"], err)
  594. if err := rc.Close(); err != nil {
  595. log.Println(err)
  596. }
  597. } else {
  598. //保存es
  599. delete(savetmp, "_id")
  600. if err := rc.Close(); err != nil {
  601. log.Println(err)
  602. }
  603. //esConn := elastic.GetEsConn()
  604. //defer elastic.DestoryEsConn(esConn)
  605. if _, err := EsConn.Index().Index(Config["elasticsearch_buyer_index"]).Type(Config["elasticsearch_buyer_type"]).Id(_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
  606. log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
  607. } else {
  608. //删除临时表
  609. FClient.DbName = Config["mgodb_extract_kf"]
  610. if deleteNum := FClient.DeleteById("buyer_new", tmpId); deleteNum == 0 {
  611. log.Println("删除临时表失败", deleteNum)
  612. }
  613. }
  614. }
  615. } else {
  616. log.Println("save mongo err:", saveid, tmp["_id"])
  617. }
  618. }
  619. }
  620. }
  621. }
  622. t2.Reset(time.Minute)
  623. }
  624. }