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