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