timedTaskWinner.go 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501
  1. package main
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "github.com/garyburd/redigo/redis"
  6. "go.mongodb.org/mongo-driver/bson/primitive"
  7. "gopkg.in/mgo.v2/bson"
  8. "log"
  9. mu "mfw/util"
  10. "net"
  11. "qfw/util"
  12. "sort"
  13. "strings"
  14. "time"
  15. )
  16. func TaskWinner(mapinfo *map[string]interface{}) {
  17. defer util.Catch()
  18. gtid, lteid := util.ObjToString((*mapinfo)["gtid"]), util.ObjToString((*mapinfo)["lteid"])
  19. if gtid == "" || lteid == "" {
  20. log.Println(gtid, lteid, "参数错误")
  21. return
  22. }
  23. GId, err := primitive.ObjectIDFromHex(gtid)
  24. LtId, err2 := primitive.ObjectIDFromHex(lteid)
  25. if err != nil || err2 != nil {
  26. log.Println(gtid, lteid, "转换_id错误")
  27. return
  28. }
  29. //udp的id区间查询bidding 中标人 中标联系人 中标联系电话
  30. // topscopeclass项目类型-industry行业类型&&topscopeclass联系人项目类型
  31. // (area地区-province省份 city城市-city城市 district区县-district区县)
  32. // winneraddr-company_address企业地址
  33. SourceClientcc := SourceClient.GetMgoConn()
  34. defer SourceClient.DestoryMongoConn(SourceClientcc)
  35. cursor := SourceClientcc.DB(Config["mgodb_bidding"]).C(Config["mgodb_mgoinit_c"]).Find(bson.M{
  36. "_id": bson.M{
  37. "$gte": GId,
  38. "$lte": LtId,
  39. },
  40. }).Select(bson.M{"winner": 1, "winnertel": 1, "winnerperson": 1,
  41. "topscopeclass": 1, "winneraddr": 1}).Iter()
  42. if cursor == nil {
  43. log.Println(cursor)
  44. return
  45. }
  46. overid := gtid
  47. tmp := map[string]interface{}{}
  48. for cursor.Next(&tmp) {
  49. overid = tmp["_id"].(primitive.ObjectID).Hex()
  50. log.Println(tmp["_id"])
  51. if tmp["winner"] == nil || tmp["winner"] == "" {
  52. continue
  53. }
  54. //redis查询是否存在
  55. rdb := RedisPool.Get()
  56. if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
  57. //redis不存在存到临时表,定时任务处理
  58. FClient.DbName = Config["mgodb_extract_kf"]
  59. if tmpid := FClient.Save("winner_new", tmp) ;tmpid==nil{
  60. log.Println("FClient.Save err",tmpid)
  61. }
  62. //log.Println("get redis id err:定时任务处理", err, tmp)
  63. if err := rdb.Close(); err != nil {
  64. log.Println(err)
  65. }
  66. continue
  67. } else {
  68. //log.Println("redis get :", reply)
  69. //redis存在
  70. //log.Println(reply)
  71. //reply = "5e0316b998a9abaf6535df3d"
  72. //id, err := primitive.ObjectIDFromHex(reply)
  73. //if err != nil {
  74. // log.Println("get redis id Hex err:", err, tmp)
  75. // if err := rdb.Close(); err != nil {
  76. // log.Println(err)
  77. // }
  78. // continue
  79. //}
  80. if err := rdb.Close(); err != nil {
  81. log.Println(err)
  82. }
  83. //拿到合并后的qyk
  84. FClient.DbName = Config["mgodb_extract_kf"]
  85. oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
  86. if oldTmp == nil{
  87. log.Println("redis id 不存在")
  88. continue
  89. }
  90. //err = FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
  91. // FindOne(context.TODO(), bson.M{"_id": id}).Decode(&oldTmp)
  92. //if err != nil {
  93. // log.Println("qyk id err:", err, id)
  94. // continue
  95. //}
  96. //比较合并
  97. //行业类型
  98. tmpTopscopeclass := []string{}
  99. tmpTopscopeclassMap := make(map[string]bool)
  100. if oldTmp["industry"] == nil {
  101. //log.Println(reflect.ValueOf(tmp["topscopeclass"]))
  102. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  103. for _, vv := range v {
  104. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  105. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  106. }
  107. }
  108. for k := range tmpTopscopeclassMap {
  109. tmpTopscopeclass = append(tmpTopscopeclass, k)
  110. }
  111. }
  112. } else {
  113. if v, ok := oldTmp["industry"].(primitive.A); ok {
  114. for _, vv := range v {
  115. if vvv, ok := vv.(string); ok {
  116. tmpTopscopeclassMap[vvv] = true
  117. }
  118. }
  119. }
  120. if v, ok := tmp["topscopeclass"].(primitive.A); ok {
  121. for _, vv := range v {
  122. if vvv, ok := vv.(string); ok && len(vvv) > 1 {
  123. tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
  124. }
  125. }
  126. for k := range tmpTopscopeclassMap {
  127. tmpTopscopeclass = append(tmpTopscopeclass, k)
  128. }
  129. }
  130. }
  131. sort.Strings(tmpTopscopeclass)
  132. oldTmp["industry"] = tmpTopscopeclass
  133. esId := oldTmp["_id"].(primitive.ObjectID).Hex()
  134. //更新行业类型
  135. if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
  136. oldTmp["updatatime"] = time.Now().Unix()
  137. //mongo更新
  138. FClient.DbName =Config["mgodb_extract_kf"]
  139. if !FClient.UpdateById(Config["mgo_qyk_c"],esId,bson.M{"$set": oldTmp}){
  140. log.Println("mongo更新err",esId)
  141. }
  142. //if _, err := FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
  143. // UpdateOne(context.TODO(), bson.M{"_id": oldTmp["_id"]}, bson.M{"$set": oldTmp}); err != nil {
  144. // log.Println("mongo更新err:", err)
  145. //}
  146. //es更新
  147. delete(oldTmp, "_id")
  148. //esConn := elastic.GetEsConn()
  149. //defer elastic.DestoryEsConn(esConn)
  150. if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  151. log.Println("update es err:", err)
  152. }
  153. //log.Println( err2,err3)
  154. continue
  155. }
  156. //联系方式合并
  157. var tmpperson, winnertel string
  158. tmpperson = tmp["winnerperson"].(string)
  159. if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
  160. winnertel = ""
  161. } else {
  162. if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
  163. winnertel = ""
  164. } else {
  165. winnertel = util.ObjToString(tmp["winnertel"])
  166. }
  167. }
  168. contactMaps := make([]interface{}, 0)
  169. if oldTmp["contact"] == nil {
  170. tmpContact := make(map[string]interface{})
  171. tmpContact["contact_person"] = tmpperson
  172. tmpContact["contact_type"] = "项目联系人"
  173. tmpContact["phone"] = winnertel
  174. tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  175. tmpContact["updatetime"] = time.Now().Unix()
  176. contactMaps = append(contactMaps, tmpContact)
  177. } else {
  178. //对比前四项,相等丢弃
  179. if v, ok := oldTmp["contact"].(primitive.A); ok {
  180. var isNotUpdate bool
  181. for _, vv := range v {
  182. if vvv, ok := vv.(map[string]interface{}); ok {
  183. if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
  184. vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
  185. isNotUpdate = true
  186. vvv["updatetime"] = time.Now().Unix()
  187. }
  188. contactMaps = append(contactMaps, vvv)
  189. }
  190. }
  191. if !isNotUpdate {
  192. vvv := make(map[string]interface{})
  193. vvv["contact_person"] = tmp["winnerperson"]
  194. vvv["contact_type"] = "项目联系人"
  195. vvv["phone"] = winnertel
  196. vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
  197. vvv["updatetime"] = time.Now().Unix()
  198. contactMaps = append(contactMaps, vvv)
  199. }
  200. }
  201. }
  202. oldTmp["contact"] = contactMaps
  203. //mongo更新
  204. oldTmp["updatatime"] = time.Now().Unix()
  205. FClient.DbName=Config["mgodb_extract_kf"]
  206. if !FClient.UpdateById(Config["mgo_qyk_c"],esId,bson.M{"$set": oldTmp}){
  207. log.Println("mongo更新 err",esId,oldTmp)
  208. }
  209. //if _, err := FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
  210. // UpdateOne(context.TODO(), bson.M{"_id": oldTmp["_id"]}, bson.M{"$set": oldTmp}); err != nil {
  211. // log.Println("mongo更新 err :", err)
  212. //}
  213. //es更新
  214. delete(oldTmp, "_id")
  215. //esConn := elastic.GetEsConn()
  216. //defer elastic.DestoryEsConn(esConn)
  217. if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
  218. log.Println("EsConn err :", err)
  219. }
  220. //log.Println( err2,err3)
  221. }
  222. }
  223. //defer cursor.Close(context.Background())
  224. //log.Println("合并执行完成", gtid, lteid, overid)
  225. //if overid != lteid {
  226. // by, _ := json.Marshal(map[string]interface{}{
  227. // "gtid": overid,
  228. // "lteid": lteid,
  229. // "stype": "",
  230. // })
  231. // if e := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
  232. // IP: net.ParseIP("127.0.0.1"),
  233. // Port: Updport,
  234. // }); e != nil {
  235. // log.Println(e)
  236. // }
  237. // log.Println("重新发送udp:", string(by))
  238. // return
  239. //}
  240. log.Println("合并执行完成 ok", gtid, lteid, overid)
  241. }
  242. //定时任务
  243. //1.存异常表
  244. //2.合并原始库新增
  245. func TimedTaskWinner() {
  246. //time.Sleep(time.Hour*70)
  247. t2 := time.NewTimer(time.Second * 5)
  248. for range t2.C {
  249. Fcconn := FClient.GetMgoConn()
  250. defer FClient.DestoryMongoConn(Fcconn)
  251. tmpLast := map[string]interface{}{}
  252. if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
  253. if !iter.Next(&tmpLast) {
  254. //临时表无数据
  255. log.Println("临时表无数据:")
  256. t2.Reset(time.Minute * 5)
  257. continue
  258. } else {
  259. log.Println("临时表有数据:", tmpLast)
  260. fconn := FClient.GetMgoConn()
  261. defer FClient.DestoryMongoConn(fconn)
  262. cursor := fconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{
  263. "_id": bson.M{
  264. "$lte": tmpLast["_id"],
  265. },
  266. }).Sort("_id").Iter()
  267. if cursor == nil {
  268. log.Println("查询失败")
  269. t2.Reset(time.Second * 5)
  270. continue
  271. }
  272. //遍历临时表数据,匹配不到原始库存入异常表
  273. tmp := make(map[string]interface{})
  274. for cursor.Next(&tmp) {
  275. tmpId := tmp["_id"].(primitive.ObjectID).Hex()
  276. //再重新查找redis,存在发udp处理,不存在走新增合并
  277. rdb := RedisPool.Get()
  278. if _, err := redis.String(rdb.Do("GET", tmp["winner"])); err == nil {
  279. //{"gtid":"57d7ad2f61a0721f152d2ad5","lteid":"5e20968d85a9271abf0ad6c2","stype":""}
  280. //redis存在发送udp进行处理
  281. by, _ := json.Marshal(map[string]interface{}{
  282. "gtid": tmpId,
  283. "lteid": tmpId,
  284. "stype": "",
  285. })
  286. if e := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
  287. IP: net.ParseIP("127.0.0.1"),
  288. Port: Updport,
  289. }); e != nil {
  290. log.Println(e)
  291. }
  292. //存在的话删除tmp mongo表
  293. FClient.DbName = Config["mgodb_extract_kf"]
  294. if DeletedCount := FClient.DeleteById("winner_new", tmpId); DeletedCount == 0 {
  295. log.Println("删除临时表err:", DeletedCount)
  296. }
  297. if err := rdb.Close(); err != nil {
  298. log.Println(err)
  299. }
  300. continue
  301. } else {
  302. if err = rdb.Close(); err != nil {
  303. log.Println(err)
  304. }
  305. }
  306. //查询redis不存在新增
  307. FClient.DbName = Config["mgodb_enterprise"]
  308. resulttmp := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": tmp["winner"]})
  309. if resulttmp["_id"] == nil {
  310. //log.Println(r)
  311. //匹配不到原始库,存入异常表删除临时表
  312. FClient.DbName = Config["mgodb_extract_kf"]
  313. if saveid := FClient.Save("winner_err", tmp); saveid == nil {
  314. log.Println("存入异常表错误", tmp)
  315. }
  316. FClient.DbName = Config["mgodb_extract_kf"]
  317. if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
  318. log.Println("删除临时表错误", deleteNum)
  319. }
  320. continue
  321. } else {
  322. //log.Println(123)
  323. //匹配到原始库,新增 resulttmp
  324. if resulttmp["credit_no"] != nil {
  325. if credit_no, ok := resulttmp["credit_no"].(string); ok && strings.TrimSpace(credit_no) != "" &&
  326. len(strings.TrimSpace(credit_no)) > 8 {
  327. dataNo := strings.TrimSpace(credit_no)[2:8]
  328. if Addrs[dataNo] != nil {
  329. if v, ok := Addrs[dataNo].(map[string]interface{}); ok {
  330. if resulttmp["province"] == nil || resulttmp["province"] == "" {
  331. resulttmp["province"] = v["province"]
  332. }
  333. resulttmp["city"] = v["city"]
  334. resulttmp["district"] = v["district"]
  335. }
  336. }
  337. }
  338. }
  339. contacts := make([]map[string]interface{}, 0)
  340. contact := make(map[string]interface{}, 0)
  341. if resulttmp["legal_person"] != nil {
  342. contact["contact_person"] = resulttmp["legal_person"] //联系人
  343. } else {
  344. contact["contact_person"] = "" //联系人
  345. }
  346. contact["contact_type"] = "法定代表人" //法定代表人
  347. //log.Println(1)
  348. if resulttmp["annual_reports"] != nil {
  349. bytes, err := json.Marshal(resulttmp["annual_reports"])
  350. if err != nil {
  351. log.Println("annual_reports err:", err)
  352. }
  353. phonetmp := make([]map[string]interface{}, 0)
  354. err = json.Unmarshal(bytes, &phonetmp)
  355. if err != nil {
  356. log.Println("Unmarshal err:", err)
  357. }
  358. for _, vv := range phonetmp {
  359. if vv["company_phone"] != nil {
  360. if vv["company_phone"] == "" {
  361. continue
  362. } else {
  363. contact["phone"] = vv["company_phone"] //联系电话
  364. break
  365. }
  366. } else {
  367. contact["phone"] = "" //联系电话
  368. }
  369. }
  370. }
  371. //log.Println(k, contact["phone"], resulttmp["_id"])
  372. //time.Sleep(10 * time.Second)
  373. if contact["phone"] == nil {
  374. contact["phone"] = "" //联系电话
  375. }
  376. contact["topscopeclass"] = "企业公示" //项目类型
  377. contact["updatetime"] = time.Now().Unix() //更新时间
  378. contacts = append(contacts, contact)
  379. resulttmp["contact"] = contacts
  380. savetmp := make(map[string]interface{}, 0)
  381. for _, sk := range Fields {
  382. if sk == "establish_date" {
  383. if resulttmp[sk] != nil {
  384. savetmp[sk] = resulttmp[sk].(primitive.DateTime).Time().UTC().Unix()
  385. continue
  386. }
  387. } else if sk == "capital" {
  388. //log.Println(sk, resulttmp[sk])
  389. savetmp[sk] = ObjToMoney([]interface{}{resulttmp[sk], ""})[0]
  390. continue
  391. } else if sk == "partners" {
  392. //log.Println(sk, resulttmp[sk], )
  393. //fmt.Println(reflect.TypeOf(resulttmp[sk]))
  394. if resulttmp[sk] != nil {
  395. if ppms, ok := resulttmp[sk].(primitive.A); ok {
  396. for i, _ := range ppms {
  397. if ppms[i].(map[string]interface{})["stock_type"] != nil {
  398. ppms[i].(map[string]interface{})["stock_type"] = "企业公示"
  399. }
  400. delete(ppms[i].(map[string]interface{}), "identify_type")
  401. }
  402. savetmp[sk] = ppms
  403. }
  404. } else {
  405. savetmp[sk] = []interface{}{}
  406. }
  407. continue
  408. } else if sk == "_id" {
  409. savetmp["tmp"+sk] = resulttmp[sk]
  410. continue
  411. } else if sk == "area_code" {
  412. //行政区划代码
  413. savetmp[sk] = fmt.Sprint(resulttmp[sk])
  414. continue
  415. } else if sk == "report_websites" {
  416. //网址
  417. if resulttmp["report_websites"] == nil {
  418. savetmp["website"] = ""
  419. } else {
  420. report_websitesArr := []string{}
  421. if ppms, ok := resulttmp[sk].(primitive.A); ok {
  422. for _, v := range ppms {
  423. if vvv, ok := v.(map[string]interface{}); ok {
  424. if rv, ok := vvv["website_url"].(string); ok {
  425. report_websitesArr = append(report_websitesArr, rv)
  426. }
  427. }
  428. }
  429. }
  430. sort.Strings(report_websitesArr)
  431. savetmp["website"] = strings.Join(report_websitesArr, ";")
  432. }
  433. continue
  434. } else if sk == "wechat_accounts" {
  435. savetmp[sk] = []interface{}{}
  436. continue
  437. }
  438. if resulttmp[sk] == nil && sk != "history_name" && sk != "wechat_accounts" && sk != "establish_date" && sk != "capital" && sk != "partners" && sk != "contact" && sk != "report_websites" {
  439. savetmp[sk] = ""
  440. } else {
  441. savetmp[sk] = resulttmp[sk]
  442. }
  443. }
  444. //tmps = append(tmps, savetmp)
  445. savetmp["updatatime"] = time.Now().Unix()
  446. //保存mongo
  447. FClient.DbName = Config["mgodb_extract_kf"]
  448. saveid := FClient.Save(Config["mgo_qyk_c"], savetmp)
  449. if saveid != nil {
  450. //保存redis
  451. rc := RedisPool.Get()
  452. var _id string
  453. if v, ok := saveid.(primitive.ObjectID); ok {
  454. _id = v.Hex()
  455. }
  456. if _, err := rc.Do("SET", savetmp["company_name"], _id); err != nil {
  457. log.Println("save redis err:", tmp["_id"], savetmp["_id"], savetmp["company_name"], err)
  458. if err := rc.Close(); err != nil {
  459. log.Println(err)
  460. }
  461. } else {
  462. //保存es
  463. delete(savetmp, "_id")
  464. if err := rc.Close(); err != nil {
  465. log.Println(err)
  466. }
  467. //esConn := elastic.GetEsConn()
  468. //defer elastic.DestoryEsConn(esConn)
  469. if _, err := EsConn.Index().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
  470. log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
  471. } else {
  472. //删除临时表
  473. FClient.DbName = Config["mgodb_extract_kf"]
  474. if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
  475. log.Println("删除临时表失败", deleteNum)
  476. }
  477. }
  478. }
  479. } else {
  480. log.Println("save mongo err:", saveid, tmp["_id"])
  481. }
  482. }
  483. }
  484. }
  485. }
  486. t2.Reset(time.Minute)
  487. }
  488. }