timedTaskWinner.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641
  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 TaskWinner(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{"winner": 1, "winnertel": 1, "winnerperson": 1,
  42. "topscopeclass": 1, "winneraddr": 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["winner"] == nil || tmp["winner"] == "" {
  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["winner"])); err != nil {
  84. //redis不存在,存到临时表,定时任务处理
  85. FClient.DbName = Config["mgodb_extract_kf"]
  86. if tmpid := FClient.Save("winner_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["winner"])
  105. continue
  106. }
  107. tmpTopscopeclass := []string{}
  108. tmpTopscopeclassMap := make(map[string]bool)
  109. if oldTmp["industry"] == nil {
  110. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  111. for _, vv := range v {
  112. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  113. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  114. }
  115. }
  116. for k := range tmpTopscopeclassMap {
  117. tmpTopscopeclass = append(tmpTopscopeclass, k)
  118. }
  119. }
  120. } else {
  121. if v, ok := oldTmp["industry"].(primitive.A); ok {
  122. for _, vv := range v {
  123. if vvv, ok := vv.(string); ok {
  124. tmpTopscopeclassMap[vvv] = true
  125. }
  126. }
  127. }
  128. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  129. for _, vv := range v {
  130. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  131. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  132. }
  133. }
  134. for k := range tmpTopscopeclassMap {
  135. tmpTopscopeclass = append(tmpTopscopeclass, k)
  136. }
  137. }
  138. }
  139. sort.Strings(tmpTopscopeclass)
  140. oldTmp["industry"] = tmpTopscopeclass
  141. esId := oldTmp["_id"].(primitive.ObjectID).Hex()
  142. //更新行业类型
  143. if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
  144. oldTmp["updatatime"] = time.Now().Unix()
  145. //mongo更新
  146. FClient.DbName = Config["mgodb_extract_kf"]
  147. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  148. log.Println("mongo更新err", esId)
  149. }
  150. //es更新
  151. delete(oldTmp, "_id")
  152. if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  153. log.Println("update es err:", err)
  154. }
  155. //删除存量redis
  156. conn.Del(redisId)
  157. continue
  158. }
  159. //联系方式合并
  160. var tmpperson, winnertel string
  161. tmpperson = tmp["winnerperson"].(string)
  162. if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
  163. winnertel = ""
  164. } else {
  165. if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
  166. winnertel = ""
  167. } else {
  168. winnertel = util.ObjToString(tmp["winnertel"])
  169. }
  170. }
  171. contactMaps := make([]interface{}, 0)
  172. if oldTmp["contact"] == nil {
  173. tmpContact := make(map[string]interface{})
  174. tmpContact["infoid"] = redisId
  175. tmpContact["contact_person"] = tmpperson
  176. tmpContact["contact_type"] = "项目联系人"
  177. tmpContact["phone"] = winnertel
  178. tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  179. tmpContact["updatetime"] = time.Now().Unix()
  180. contactMaps = append(contactMaps, tmpContact)
  181. } else {
  182. //对比前四项,相等丢弃
  183. if v, ok := oldTmp["contact"].(primitive.A); ok {
  184. var isNotUpdate bool
  185. for _, vv := range v {
  186. if vvv, ok := vv.(map[string]interface{}); ok {
  187. if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
  188. vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
  189. isNotUpdate = true
  190. vvv["updatetime"] = time.Now().Unix()
  191. }
  192. contactMaps = append(contactMaps, vvv)
  193. }
  194. }
  195. if !isNotUpdate {
  196. vvv := make(map[string]interface{})
  197. vvv["infoid"] = redisId
  198. vvv["contact_person"] = tmp["winnerperson"]
  199. vvv["contact_type"] = "项目联系人"
  200. vvv["phone"] = winnertel
  201. vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  202. vvv["updatetime"] = time.Now().Unix()
  203. contactMaps = append(contactMaps, vvv)
  204. }
  205. }
  206. }
  207. oldTmp["contact"] = contactMaps
  208. //mongo更新
  209. oldTmp["updatatime"] = time.Now().Unix()
  210. FClient.DbName = Config["mgodb_extract_kf"]
  211. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  212. log.Println("存量 mongo更新 err", esId, oldTmp)
  213. }
  214. //es更新
  215. delete(oldTmp, "_id")
  216. if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  217. log.Println("存量 EsConn err :", err)
  218. }
  219. //最后删除redis
  220. conn.Del(redisId)
  221. }
  222. }
  223. }
  224. log.Println("存量历史合并执行完成 ok", gtid, lteid)
  225. } else {
  226. //增量处理
  227. overid := gtid
  228. tmp := map[string]interface{}{}
  229. for cursor.Next(&tmp) {
  230. overid = tmp["_id"].(primitive.ObjectID).Hex()
  231. log.Println(tmp["_id"])
  232. if tmp["winner"] == nil || tmp["winner"] == "" {
  233. continue
  234. }
  235. //redis查询是否存在
  236. rdb := RedisPool.Get()
  237. rdb.Do("SELECT","1")
  238. if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
  239. //redis不存在存到临时表,定时任务处理
  240. FClient.DbName = Config["mgodb_extract_kf"]
  241. if tmpid := FClient.Save("winner_new", tmp); tmpid == nil {
  242. log.Println("FClient.Save err", tmpid)
  243. }
  244. //log.Println("get redis id err:定时任务处理", err, tmp)
  245. if err := rdb.Close(); err != nil {
  246. log.Println(err)
  247. }
  248. continue
  249. } else {
  250. if err := rdb.Close(); err != nil {
  251. log.Println(err)
  252. }
  253. //拿到合并后的qyk
  254. FClient.DbName = Config["mgodb_extract_kf"]
  255. oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
  256. if oldTmp == nil {
  257. log.Println("redis id 不存在")
  258. continue
  259. }
  260. //比较合并
  261. //行业类型
  262. tmpTopscopeclass := []string{}
  263. tmpTopscopeclassMap := make(map[string]bool)
  264. if oldTmp["industry"] == nil {
  265. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  266. for _, vv := range v {
  267. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  268. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  269. }
  270. }
  271. for k := range tmpTopscopeclassMap {
  272. tmpTopscopeclass = append(tmpTopscopeclass, k)
  273. }
  274. }
  275. } else {
  276. if v, ok := oldTmp["industry"].(primitive.A); ok {
  277. for _, vv := range v {
  278. if vvv, ok := vv.(string); ok {
  279. tmpTopscopeclassMap[vvv] = true
  280. }
  281. }
  282. }
  283. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  284. for _, vv := range v {
  285. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  286. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  287. }
  288. }
  289. for k := range tmpTopscopeclassMap {
  290. tmpTopscopeclass = append(tmpTopscopeclass, k)
  291. }
  292. }
  293. }
  294. sort.Strings(tmpTopscopeclass)
  295. oldTmp["industry"] = tmpTopscopeclass
  296. esId := oldTmp["_id"].(primitive.ObjectID).Hex()
  297. //更新行业类型
  298. if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
  299. oldTmp["updatatime"] = time.Now().Unix()
  300. //mongo更新
  301. FClient.DbName = Config["mgodb_extract_kf"]
  302. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  303. log.Println("mongo更新err", esId)
  304. }
  305. //es更新
  306. delete(oldTmp, "_id")
  307. if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  308. log.Println("update es err:", err)
  309. }
  310. continue
  311. }
  312. //联系方式合并
  313. var tmpperson, winnertel string
  314. tmpperson = tmp["winnerperson"].(string)
  315. if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
  316. winnertel = ""
  317. } else {
  318. if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
  319. winnertel = ""
  320. } else {
  321. winnertel = util.ObjToString(tmp["winnertel"])
  322. }
  323. }
  324. contactMaps := make([]interface{}, 0)
  325. if oldTmp["contact"] == nil {
  326. tmpContact := make(map[string]interface{})
  327. tmpContact["infoid"] = overid
  328. tmpContact["contact_person"] = tmpperson
  329. tmpContact["contact_type"] = "项目联系人"
  330. tmpContact["phone"] = winnertel
  331. tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  332. tmpContact["updatetime"] = time.Now().Unix()
  333. contactMaps = append(contactMaps, tmpContact)
  334. } else {
  335. //对比前四项,相等丢弃
  336. if v, ok := oldTmp["contact"].(primitive.A); ok {
  337. var isNotUpdate bool
  338. for _, vv := range v {
  339. if vvv, ok := vv.(map[string]interface{}); ok {
  340. if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
  341. vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
  342. isNotUpdate = true
  343. vvv["updatetime"] = time.Now().Unix()
  344. }
  345. contactMaps = append(contactMaps, vvv)
  346. }
  347. }
  348. if !isNotUpdate {
  349. vvv := make(map[string]interface{})
  350. vvv["infoid"] = overid
  351. vvv["contact_person"] = tmp["winnerperson"]
  352. vvv["contact_type"] = "项目联系人"
  353. vvv["phone"] = winnertel
  354. vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  355. vvv["updatetime"] = time.Now().Unix()
  356. contactMaps = append(contactMaps, vvv)
  357. }
  358. }
  359. }
  360. oldTmp["contact"] = contactMaps
  361. //mongo更新
  362. oldTmp["updatatime"] = time.Now().Unix()
  363. FClient.DbName = Config["mgodb_extract_kf"]
  364. if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
  365. log.Println("mongo更新 err", esId, oldTmp)
  366. }
  367. //es更新
  368. delete(oldTmp, "_id")
  369. if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  370. log.Println("EsConn err :", err)
  371. }
  372. }
  373. }
  374. log.Println("增量合并执行完成 ok", gtid, lteid, overid)
  375. }
  376. }
  377. //定时任务 新增
  378. //1.存异常表
  379. //2.合并原始库新增
  380. func TimedTaskWinner() {
  381. //time.Sleep(time.Hour*70)
  382. t2 := time.NewTimer(time.Second * 5)
  383. for range t2.C {
  384. Fcconn := FClient.GetMgoConn()
  385. defer FClient.DestoryMongoConn(Fcconn)
  386. tmpLast := map[string]interface{}{}
  387. if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
  388. if !iter.Next(&tmpLast) {
  389. //临时表无数据
  390. log.Println("临时表无数据:")
  391. t2.Reset(time.Minute * 5)
  392. continue
  393. } else {
  394. log.Println("临时表有数据:", tmpLast)
  395. fconn := FClient.GetMgoConn()
  396. defer FClient.DestoryMongoConn(fconn)
  397. cursor := fconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{
  398. "_id": bson.M{
  399. "$lte": tmpLast["_id"],
  400. },
  401. }).Sort("_id").Iter()
  402. if cursor == nil {
  403. log.Println("查询失败")
  404. t2.Reset(time.Second * 5)
  405. continue
  406. }
  407. //遍历临时表数据,匹配不到原始库存入异常表
  408. tmp := make(map[string]interface{})
  409. for cursor.Next(&tmp) {
  410. tmpId := tmp["_id"].(primitive.ObjectID).Hex()
  411. //再重新查找redis,存在发udp处理,不存在走新增合并
  412. rdb := RedisPool.Get()
  413. rdb.Do("SELECT","1")
  414. if _, err := redis.String(rdb.Do("GET", tmp["winner"])); err == nil {
  415. //{"gtid":"57d7ad2f61a0721f152d2ad5","lteid":"5e20968d85a9271abf0ad6c2","stype":""}
  416. //redis存在发送udp进行处理
  417. by, _ := json.Marshal(map[string]interface{}{
  418. "gtid": tmpId,
  419. "lteid": tmpId,
  420. "stype": "",
  421. })
  422. if e := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
  423. IP: net.ParseIP("127.0.0.1"),
  424. Port: Updport,
  425. }); e != nil {
  426. log.Println(e)
  427. }
  428. //存在的话删除tmp mongo表
  429. FClient.DbName = Config["mgodb_extract_kf"]
  430. if DeletedCount := FClient.DeleteById("winner_new", tmpId); DeletedCount == 0 {
  431. log.Println("删除临时表err:", DeletedCount)
  432. }
  433. if err := rdb.Close(); err != nil {
  434. log.Println(err)
  435. }
  436. continue
  437. } else {
  438. if err = rdb.Close(); err != nil {
  439. log.Println(err)
  440. }
  441. }
  442. //查询redis不存在新增
  443. FClient.DbName = Config["mgodb_enterprise"]
  444. resulttmp := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": tmp["winner"]})
  445. if resulttmp["_id"] == nil {
  446. //log.Println(r)
  447. //匹配不到原始库,存入异常表删除临时表
  448. FClient.DbName = Config["mgodb_extract_kf"]
  449. if saveid := FClient.Save("winner_err", tmp); saveid == nil {
  450. log.Println("存入异常表错误", tmp)
  451. }
  452. FClient.DbName = Config["mgodb_extract_kf"]
  453. if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
  454. log.Println("删除临时表错误", deleteNum)
  455. }
  456. continue
  457. } else {
  458. //log.Println(123)
  459. //匹配到原始库,新增 resulttmp
  460. if resulttmp["credit_no"] != nil {
  461. if credit_no, ok := resulttmp["credit_no"].(string); ok && strings.TrimSpace(credit_no) != "" &&
  462. len(strings.TrimSpace(credit_no)) > 8 {
  463. dataNo := strings.TrimSpace(credit_no)[2:8]
  464. if Addrs[dataNo] != nil {
  465. if v, ok := Addrs[dataNo].(map[string]interface{}); ok {
  466. if resulttmp["province"] == nil || resulttmp["province"] == "" {
  467. resulttmp["province"] = v["province"]
  468. }
  469. resulttmp["city"] = v["city"]
  470. resulttmp["district"] = v["district"]
  471. }
  472. }
  473. }
  474. }
  475. contacts := make([]map[string]interface{}, 0)
  476. contact := make(map[string]interface{}, 0)
  477. if resulttmp["legal_person"] != nil {
  478. contact["contact_person"] = resulttmp["legal_person"] //联系人
  479. } else {
  480. contact["contact_person"] = "" //联系人
  481. }
  482. contact["contact_type"] = "法定代表人" //法定代表人
  483. //log.Println(1)
  484. if resulttmp["annual_reports"] != nil {
  485. bytes, err := json.Marshal(resulttmp["annual_reports"])
  486. if err != nil {
  487. log.Println("annual_reports err:", err)
  488. }
  489. phonetmp := make([]map[string]interface{}, 0)
  490. err = json.Unmarshal(bytes, &phonetmp)
  491. if err != nil {
  492. log.Println("Unmarshal err:", err)
  493. }
  494. for _, vv := range phonetmp {
  495. if vv["company_phone"] != nil {
  496. if vv["company_phone"] == "" {
  497. continue
  498. } else {
  499. contact["phone"] = vv["company_phone"] //联系电话
  500. break
  501. }
  502. } else {
  503. contact["phone"] = "" //联系电话
  504. }
  505. }
  506. }
  507. //log.Println(k, contact["phone"], resulttmp["_id"])
  508. //time.Sleep(10 * time.Second)
  509. if contact["phone"] == nil {
  510. contact["phone"] = "" //联系电话
  511. }
  512. contact["topscopeclass"] = "企业公示" //项目类型
  513. contact["updatetime"] = time.Now().Unix() //更新时间
  514. contact["infoid"] = "" //招标信息id
  515. contacts = append(contacts, contact)
  516. resulttmp["contact"] = contacts
  517. savetmp := make(map[string]interface{}, 0)
  518. for _, sk := range Fields {
  519. if sk == "establish_date" {
  520. if resulttmp[sk] != nil {
  521. savetmp[sk] = resulttmp[sk].(primitive.DateTime).Time().UTC().Unix()
  522. continue
  523. }
  524. } else if sk == "capital" {
  525. //log.Println(sk, resulttmp[sk])
  526. savetmp[sk] = ObjToMoney([]interface{}{resulttmp[sk], ""})[0]
  527. continue
  528. } else if sk == "partners" {
  529. //log.Println(sk, resulttmp[sk], )
  530. //fmt.Println(reflect.TypeOf(resulttmp[sk]))
  531. if resulttmp[sk] != nil {
  532. if ppms, ok := resulttmp[sk].(primitive.A); ok {
  533. for i, _ := range ppms {
  534. if ppms[i].(map[string]interface{})["stock_type"] != nil {
  535. ppms[i].(map[string]interface{})["stock_type"] = "企业公示"
  536. }
  537. delete(ppms[i].(map[string]interface{}), "identify_type")
  538. }
  539. savetmp[sk] = ppms
  540. }
  541. } else {
  542. savetmp[sk] = []interface{}{}
  543. }
  544. continue
  545. } else if sk == "_id" {
  546. savetmp["tmp"+sk] = resulttmp[sk]
  547. continue
  548. } else if sk == "area_code" {
  549. //行政区划代码
  550. savetmp[sk] = fmt.Sprint(resulttmp[sk])
  551. continue
  552. } else if sk == "report_websites" {
  553. //网址
  554. if resulttmp["report_websites"] == nil {
  555. savetmp["website"] = ""
  556. } else {
  557. report_websitesArr := []string{}
  558. if ppms, ok := resulttmp[sk].(primitive.A); ok {
  559. for _, v := range ppms {
  560. if vvv, ok := v.(map[string]interface{}); ok {
  561. if rv, ok := vvv["website_url"].(string); ok {
  562. report_websitesArr = append(report_websitesArr, rv)
  563. }
  564. }
  565. }
  566. }
  567. sort.Strings(report_websitesArr)
  568. savetmp["website"] = strings.Join(report_websitesArr, ";")
  569. }
  570. continue
  571. } else if sk == "wechat_accounts" {
  572. savetmp[sk] = []interface{}{}
  573. continue
  574. }
  575. if resulttmp[sk] == nil && sk != "history_name" && sk != "wechat_accounts" && sk != "establish_date" && sk != "capital" && sk != "partners" && sk != "contact" && sk != "report_websites" {
  576. savetmp[sk] = ""
  577. } else {
  578. savetmp[sk] = resulttmp[sk]
  579. }
  580. }
  581. //tmps = append(tmps, savetmp)
  582. savetmp["updatatime"] = time.Now().Unix()
  583. //保存mongo
  584. FClient.DbName = Config["mgodb_extract_kf"]
  585. saveid := FClient.Save(Config["mgo_qyk_c"], savetmp)
  586. if saveid != nil {
  587. //保存redis
  588. rc := RedisPool.Get()
  589. rc.Do("SELECT","1")
  590. var _id string
  591. if v, ok := saveid.(primitive.ObjectID); ok {
  592. _id = v.Hex()
  593. }
  594. if _, err := rc.Do("SET", savetmp["company_name"], _id); err != nil {
  595. log.Println("save redis err:", tmp["_id"], savetmp["_id"], savetmp["company_name"], err)
  596. if err := rc.Close(); err != nil {
  597. log.Println(err)
  598. }
  599. } else {
  600. //保存es
  601. delete(savetmp, "_id")
  602. if err := rc.Close(); err != nil {
  603. log.Println(err)
  604. }
  605. //esConn := elastic.GetEsConn()
  606. //defer elastic.DestoryEsConn(esConn)
  607. if _, err := EsConn.Index().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
  608. log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
  609. } else {
  610. //删除临时表
  611. FClient.DbName = Config["mgodb_extract_kf"]
  612. if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
  613. log.Println("删除临时表失败", deleteNum)
  614. }
  615. }
  616. }
  617. } else {
  618. log.Println("save mongo err:", saveid, tmp["_id"])
  619. }
  620. }
  621. }
  622. }
  623. }
  624. t2.Reset(time.Minute)
  625. }
  626. }