timedTaskWinner.go 25 KB

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