timedTaskWinner.go 24 KB

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