bidding.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452
  1. package main
  2. import (
  3. "fmt"
  4. "github.com/wcc4869/common_utils/log"
  5. "go.uber.org/zap"
  6. util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  7. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  8. "strings"
  9. )
  10. // updateBiddingBasicClassTest 测试环境更新 basicClass
  11. func updateBiddingBasicClassTest() {
  12. defer util.Catch()
  13. sess := MgoT.GetMgoConn()
  14. defer MgoT.DestoryMongoConn(sess)
  15. it := sess.DB("qfw_data").C("bidding").Find(nil).Select(nil).Sort("-_id").Iter()
  16. count := 0
  17. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  18. if count%5000 == 0 {
  19. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  20. }
  21. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  22. update := make(map[string]interface{})
  23. if _, ok := tmp["basicClass"]; ok && util.ObjToString(tmp["basicClass"]) != "" {
  24. update["basicClass"] = tmp["basicClass"]
  25. }
  26. // 更新Es 数据
  27. if len(update) > 0 {
  28. ////更新MongoDB
  29. //updatePool <- []map[string]interface{}{
  30. // {"_id": tmp["_id"]},
  31. // {"$set": update},
  32. //}
  33. // 更新es
  34. updateEsPool <- []map[string]interface{}{
  35. {"_id": biddingID},
  36. update,
  37. }
  38. }
  39. }
  40. }
  41. // updateBiddingBasicClass 更新bidding basicClass 存量数据
  42. func updateBiddingBasicClass() {
  43. defer util.Catch()
  44. sess := MgoB.GetMgoConn()
  45. defer MgoB.DestoryMongoConn(sess)
  46. where := map[string]interface{}{
  47. "comeintime": map[string]interface{}{
  48. "$lte": 1751273717,
  49. },
  50. }
  51. it := sess.DB("qfw").C("bidding").Find(where).Select(nil).Sort("-_id").Iter()
  52. count := 0
  53. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  54. if count%5000 == 0 {
  55. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  56. }
  57. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  58. update := make(map[string]interface{})
  59. if _, ok := tmp["basicClass"]; ok && util.ObjToString(tmp["basicClass"]) != "" {
  60. update["basicClass"] = tmp["basicClass"]
  61. }
  62. // 更新Es 数据
  63. if len(update) > 0 {
  64. ////更新MongoDB
  65. //updatePool <- []map[string]interface{}{
  66. // {"_id": tmp["_id"]},
  67. // {"$set": update},
  68. //}
  69. // 更新es
  70. updateEsPool <- []map[string]interface{}{
  71. {"_id": biddingID},
  72. update,
  73. }
  74. }
  75. }
  76. }
  77. // updateBiddingType22 根据爬虫代码,更新标讯分类
  78. func updateBiddingTypeBySpidecode() {
  79. defer util.Catch()
  80. sess := MgoB.GetMgoConn()
  81. defer MgoB.DestoryMongoConn(sess)
  82. where := map[string]interface{}{
  83. "comeintime": map[string]interface{}{
  84. "$gte": 1735660800,
  85. //"$lte": 1750908768,
  86. "$lte": 1751438445,
  87. },
  88. }
  89. it := sess.DB("qfw").C("bidding").Find(where).Select(nil).Iter()
  90. count := 0
  91. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  92. if count%1000 == 0 {
  93. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  94. }
  95. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  96. spidecode := util.ObjToString(tmp["spidercode"])
  97. site := util.ObjToString(tmp["site"])
  98. if spidecode == "tj_tjstzxmzxspbsdt_sybj" && site == "天津市投资项目在线审批办事大厅" {
  99. update := make(map[string]interface{})
  100. update["toptype"] = "拟建"
  101. update["subtype"] = "拟建"
  102. update["infoformat"] = 1
  103. // 更新Es 数据
  104. if len(update) > 0 {
  105. ////更新MongoDB
  106. updatePool <- []map[string]interface{}{
  107. {"_id": tmp["_id"]},
  108. {"$set": update},
  109. }
  110. // 更新es
  111. updateEsPool <- []map[string]interface{}{
  112. {"_id": biddingID},
  113. update,
  114. }
  115. }
  116. }
  117. }
  118. }
  119. func updateBiddingisValidFile() {
  120. defer util.Catch()
  121. sess := MgoB.GetMgoConn()
  122. defer MgoB.DestoryMongoConn(sess)
  123. where := map[string]interface{}{
  124. "comeintime": map[string]interface{}{
  125. "$gte": 1750755600,
  126. "$lte": 1750822200,
  127. },
  128. }
  129. it := sess.DB("qfw").C("bidding").Find(where).Select(nil).Sort("-_id").Iter()
  130. count := 0
  131. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  132. if count%1000 == 0 {
  133. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  134. }
  135. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  136. update := make(map[string]interface{})
  137. if _, ok := tmp["isValidFile"]; ok {
  138. update["isValidFile"] = tmp["isValidFile"]
  139. }
  140. // 更新Es 数据
  141. if len(update) > 0 {
  142. ////更新MongoDB
  143. //updatePool <- []map[string]interface{}{
  144. // {"_id": tmp["_id"]},
  145. // {"$set": update},
  146. //}
  147. // 更新es
  148. updateEsPool <- []map[string]interface{}{
  149. {"_id": biddingID},
  150. update,
  151. }
  152. }
  153. }
  154. }
  155. // updateBiddingType 更新bidding 分类类型
  156. func updateBiddingType() {
  157. defer util.Catch()
  158. sess := MgoB.GetMgoConn()
  159. defer MgoB.DestoryMongoConn(sess)
  160. where := map[string]interface{}{
  161. "_id": map[string]interface{}{
  162. "$gte": mongodb.StringTOBsonId("5a862f0640d2d9bbe88e3cec"),
  163. //"$lte": mongodb.StringTOBsonId("68552e15c936757aa1774910"),
  164. "$lte": mongodb.StringTOBsonId("68551ebfc936757aa176c9ae"),
  165. },
  166. }
  167. it := sess.DB("qfw").C("bidding").Find(where).Select(nil).Sort("-_id").Iter()
  168. count := 0
  169. count2 := 0
  170. ruleStr := `(开标记录表|开标一览表),(开标参与人|开标地点|开标时间|开标记录|开标一览表)3^(公开招标公告|中标人信息|供应商资格要求|投标人资格要求|招标条件|潜在投标人)`
  171. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  172. if count%1000 == 0 {
  173. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  174. }
  175. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  176. title := util.ObjToString(tmp["title"])
  177. detail := util.ObjToString(tmp["detail"])
  178. update := make(map[string]interface{})
  179. //匹配标题
  180. if strings.Contains(title, "开标记录") {
  181. update["toptype"] = "结果"
  182. update["subtype"] = "开标记录"
  183. } else {
  184. //匹配内容
  185. result := MatchAllRules(detail, ruleStr)
  186. if result.Matched {
  187. update["toptype"] = "结果"
  188. update["subtype"] = "开标记录"
  189. }
  190. }
  191. // 更新Es 数据
  192. if len(update) > 0 {
  193. count2++
  194. if count2%1000 == 0 {
  195. log.Info("updateBiddingType", zap.Int("count2", count2), zap.Any("id", biddingID))
  196. }
  197. //更新MongoDB
  198. updatePool <- []map[string]interface{}{
  199. {"_id": tmp["_id"]},
  200. {"$set": update},
  201. }
  202. // 更新es
  203. updateEsPool <- []map[string]interface{}{
  204. {"_id": biddingID},
  205. update,
  206. }
  207. }
  208. }
  209. }
  210. // dealBiddingNiJian 更新bidding ,owner 不为空的赋值给buyer
  211. //if toptype == "拟建"
  212. // if tmp["owner"] != nil
  213. // tmp["buyer"] = tmp["owner"]
  214. func dealBiddingNiJian() {
  215. defer util.Catch()
  216. sess := MgoB.GetMgoConn()
  217. defer MgoB.DestoryMongoConn(sess)
  218. where := map[string]interface{}{
  219. "comeintime": map[string]interface{}{
  220. "$lt": 1735228800,
  221. },
  222. "toptype": "拟建",
  223. }
  224. it := sess.DB("qfw").C("bidding").Find(where).Select(nil).Iter()
  225. fmt.Println("taskRun 开始")
  226. count := 0
  227. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  228. if count%10000 == 0 {
  229. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  230. }
  231. if util.ObjToString(tmp["toptype"]) == "拟建" {
  232. update := map[string]interface{}{}
  233. esUpdate := map[string]interface{}{}
  234. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  235. if tmp["owner"] != nil {
  236. update["buyer"] = tmp["owner"]
  237. esUpdate["buyer"] = tmp["owner"]
  238. }
  239. if len(update) > 0 {
  240. //更新mongo
  241. //MgoT.UpdateById("bidding", biddingID, map[string]interface{}{"$set": update})
  242. //更新MongoDB
  243. updatePool <- []map[string]interface{}{
  244. {"_id": tmp["_id"]},
  245. {"$set": update},
  246. }
  247. //2.es 项目 更新字段
  248. //err := Es.UpdateDocument("bidding", biddingID, update)
  249. //if err != nil && err.Error() != "Document not updated: noop" {
  250. // log.Info("bidding es update err", err, biddingID)
  251. //}
  252. //// 更新es
  253. //updateEsPool <- []map[string]interface{}{
  254. // {"_id": biddingID},
  255. // update,
  256. //}
  257. }
  258. // 更新Es 数据
  259. if len(esUpdate) > 0 {
  260. // 更新es
  261. updateEsPool <- []map[string]interface{}{
  262. {"_id": biddingID},
  263. esUpdate,
  264. }
  265. }
  266. }
  267. }
  268. log.Info("Run Over...Count:", log.Int("count", count))
  269. }
  270. // updateBiddingBidamount 处理bidding 中标金额,预算数据
  271. func updateBiddingBidamount() {
  272. defer util.Catch()
  273. sess := MgoB.GetMgoConn()
  274. defer MgoB.DestoryMongoConn(sess)
  275. it := sess.DB("qfw").C("zktest_repair_0107").Find(nil).Select(nil).Iter()
  276. fmt.Println("taskRun 开始")
  277. count := 0
  278. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  279. if count%10000 == 0 {
  280. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  281. }
  282. update := make(map[string]interface{})
  283. if bidamount, ok := tmp["bidamount"]; ok {
  284. update["bidamount"] = bidamount
  285. } else {
  286. update["bidamount"] = 0.0
  287. }
  288. if _, ok := tmp["budget"]; ok {
  289. update["budget"] = tmp["budget"]
  290. } else {
  291. update["budget"] = 0.0
  292. }
  293. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  294. if len(update) > 0 {
  295. //project := getProject(biddingID)
  296. //if len(project) > 0 {
  297. // //projectID := mongodb.BsonIdToSId(project["_id"])
  298. // //err := Es.UpdateDocument("projectset", projectID, update)
  299. // //err = EsNew.UpdateDocument("projectset", projectID, update)
  300. // //if err != nil && err.Error() != "Document not updated: noop" {
  301. // // log.Info("bidding es update err", err, biddingID)
  302. // //}
  303. //}
  304. // 更新bidding es
  305. err := Es.UpdateDocument("bidding", biddingID, update)
  306. err = EsNew.UpdateDocument("bidding", biddingID, update)
  307. if err != nil && err.Error() != "Document not updated: noop" {
  308. log.Info("bidding es update err", err, biddingID)
  309. }
  310. }
  311. }
  312. log.Info("数据处理完毕", log.Int("total", count))
  313. }
  314. // updateBiddingToptype 更新招标分类结果
  315. func updateBiddingToptype() {
  316. defer util.Catch()
  317. sess := MgoB.GetMgoConn()
  318. defer MgoB.DestoryMongoConn(sess)
  319. it := sess.DB("qfw").C("wcc_bidding_kaibiao").Find(nil).Select(nil).Sort("-_id").Iter()
  320. count := 0
  321. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  322. if count%1000 == 0 {
  323. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  324. }
  325. biddingID := mongodb.BsonIdToSId(tmp["_id"])
  326. update := make(map[string]interface{})
  327. subtype := util.ObjToString(tmp["subtype"])
  328. if subtype == "开标记录" {
  329. continue
  330. }
  331. update["toptype"] = tmp["toptype"]
  332. update["subtype"] = tmp["subtype"]
  333. // 更新Es 数据
  334. if len(update) > 0 {
  335. //更新MongoDB
  336. updatePool <- []map[string]interface{}{
  337. {"_id": tmp["_id"]},
  338. {"$set": update},
  339. }
  340. // 更新es
  341. updateEsPool <- []map[string]interface{}{
  342. {"_id": biddingID},
  343. update,
  344. }
  345. }
  346. }
  347. }
  348. // updateBiddingBuyer g 更新buyer
  349. func updateBiddingBuyer() {
  350. defer util.Catch()
  351. sess := MgoB.GetMgoConn()
  352. defer MgoB.DestoryMongoConn(sess)
  353. it := sess.DB("qfw").C("wcc_buyer_test").Find(nil).Select(nil).Iter()
  354. count := 0
  355. for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
  356. if count%100 == 0 {
  357. log.Info("current", log.Int("count", count), log.Any("_id", tmp["_id"]))
  358. }
  359. biddingID := util.ObjToString(tmp["id"])
  360. update := make(map[string]interface{})
  361. update["buyer"] = "海南警察学院(筹)"
  362. // 更新Es 数据
  363. if len(update) > 0 {
  364. //更新MongoDB
  365. MgoB.UpdateById("bidding", biddingID, map[string]interface{}{"$set": update})
  366. //2.es 项目 更新字段
  367. err := Es.UpdateDocument("bidding", biddingID, update)
  368. err = EsNew.UpdateDocument("bidding", biddingID, update)
  369. err = EsNew.UpdateDocument("bidding_temp", biddingID, update)
  370. if err != nil && err.Error() != "Document not updated: noop" {
  371. log.Info("bidding es update err", err, biddingID)
  372. }
  373. }
  374. }
  375. }