timedTaskBuyer.go 22 KB

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