timedTaskBuyer.go 22 KB

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