timedTaskWinner.go 22 KB

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