timedTaskBuyer.go 21 KB

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