timedTaskWinner.go 22 KB

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