init.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348
  1. package main
  2. import (
  3. "context"
  4. "esindex/config"
  5. "fmt"
  6. es7 "github.com/olivere/elastic/v7"
  7. "go.uber.org/zap"
  8. util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  9. "jygit.jydev.jianyu360.cn/data_processing/common_utils/elastic"
  10. "jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
  11. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  12. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mysqldb"
  13. "os"
  14. "strings"
  15. "time"
  16. )
  17. var (
  18. ProjectField = make(map[string]string, 500) //项目字段
  19. ProjectListF = make(map[string]string, 200)
  20. BiddingField = make(map[string]string, 200) //bidding_processing_field, level=1 最外层字段,
  21. BiddingLevelField = make(map[string]map[string]string) //level=2 的第二层字段
  22. )
  23. // InitLog @Description
  24. // @Author J 2022/7/26 15:30
  25. func InitLog() {
  26. now := time.Now()
  27. logcfg := config.Conf.Log
  28. err := log.InitLog(
  29. log.Path(logcfg.LogPath),
  30. log.Level(logcfg.LogLevel),
  31. log.Compress(logcfg.Compress),
  32. log.MaxSize(logcfg.MaxSize),
  33. log.MaxBackups(logcfg.MaxBackups),
  34. log.MaxAge(logcfg.MaxAge),
  35. log.Format(logcfg.Format),
  36. )
  37. if err != nil {
  38. fmt.Printf("InitLog failed: %v\n", err)
  39. os.Exit(1)
  40. }
  41. log.Info("InitLog", zap.Any("duration", time.Since(now).Seconds()))
  42. }
  43. func InitMgo() {
  44. now := time.Now()
  45. MgoB = &mongodb.MongodbSim{
  46. MongodbAddr: config.Conf.DB.MongoB.Addr,
  47. DbName: config.Conf.DB.MongoB.Dbname,
  48. Size: config.Conf.DB.MongoB.Size,
  49. UserName: config.Conf.DB.MongoB.User,
  50. Password: config.Conf.DB.MongoB.Password,
  51. Direct: config.Conf.DB.MongoB.Direct,
  52. }
  53. MgoB.InitPool()
  54. if config.Conf.DB.MongoB.Addr == "" || config.Conf.DB.MongoB.Dbname == "" {
  55. log.Error("InitMgo", zap.String("MgoB", "地址或者数据库为空"))
  56. }
  57. if config.Conf.DB.MongoB.Coll == "" {
  58. log.Error("InitMgo", zap.String("MgoB", "查询表为空"))
  59. }
  60. log.Info("InitMgo", zap.Any("MgoB duration", time.Since(now).Seconds()))
  61. //项目信息
  62. MgoP = &mongodb.MongodbSim{
  63. MongodbAddr: config.Conf.DB.MongoP.Addr,
  64. DbName: config.Conf.DB.MongoP.Dbname,
  65. Size: config.Conf.DB.MongoP.Size,
  66. UserName: config.Conf.DB.MongoP.User,
  67. Password: config.Conf.DB.MongoP.Password,
  68. }
  69. MgoP.InitPool()
  70. if config.Conf.DB.MongoP.Addr == "" || config.Conf.DB.MongoP.Dbname == "" {
  71. log.Error("InitMgo", zap.String("MongoP", "地址或者数据库为空"))
  72. }
  73. if config.Conf.DB.MongoP.Coll == "" {
  74. log.Error("InitMgo", zap.String("MongoP", "查询表为空"))
  75. }
  76. log.Info("InitMgo", zap.Any("MgoP duration", time.Since(now).Seconds()))
  77. //中标单位定时同步
  78. MgoQ = &mongodb.MongodbSim{
  79. MongodbAddr: config.Conf.DB.MongoQ.Addr,
  80. DbName: config.Conf.DB.MongoQ.Dbname,
  81. Size: config.Conf.DB.MongoQ.Size,
  82. UserName: config.Conf.DB.MongoQ.User,
  83. Password: config.Conf.DB.MongoQ.Password,
  84. }
  85. MgoQ.InitPool()
  86. if config.Conf.DB.MongoQ.Addr == "" || config.Conf.DB.MongoQ.Dbname == "" {
  87. log.Error("InitMgo", zap.String("MongoQ", "地址或者数据库为空"))
  88. }
  89. log.Info("InitMgo", zap.Any("MgoQ duration", time.Since(now).Seconds()))
  90. //采购单位
  91. Mysql = &mysqldb.Mysql{
  92. Address: config.Conf.DB.MysqlB.Addr,
  93. DBName: config.Conf.DB.MysqlB.Dbname,
  94. UserName: config.Conf.DB.MysqlB.Username,
  95. PassWord: config.Conf.DB.MysqlB.Password,
  96. }
  97. Mysql.Init()
  98. if config.Conf.DB.MysqlB.Addr == "" || config.Conf.DB.MysqlB.Dbname == "" {
  99. log.Error("InitMgo", zap.String("Mysql", "地址或者数据库为空"))
  100. }
  101. log.Info("InitMgo", zap.Any("MysqlB duration", time.Since(now).Seconds()))
  102. //181 特殊企业,采购单位验证
  103. MgoS = &mongodb.MongodbSim{
  104. MongodbAddr: config.Conf.DB.MongoS.Addr,
  105. DbName: config.Conf.DB.MongoS.Dbname,
  106. Size: config.Conf.DB.MongoS.Size,
  107. UserName: config.Conf.DB.MongoS.User,
  108. Password: config.Conf.DB.MongoS.Password,
  109. }
  110. MgoS.InitPool()
  111. if config.Conf.DB.MongoS.Addr == "" || config.Conf.DB.MongoS.Dbname == "" {
  112. log.Error("InitMgo", zap.String("MongoS", "地址或者数据库为空"))
  113. }
  114. log.Info("InitMgo", zap.Any("MgoS duration", time.Since(now).Seconds()))
  115. }
  116. func InitEs() {
  117. now := time.Now()
  118. Es = &elastic.Elastic{
  119. S_esurl: config.Conf.DB.Es.Addr,
  120. I_size: config.Conf.DB.Es.Size,
  121. Username: config.Conf.DB.Es.Username,
  122. Password: config.Conf.DB.Es.Password,
  123. }
  124. Es.InitElasticSize()
  125. if config.Conf.DB.Es.Addr == "" {
  126. log.Error("InitEs", zap.String("ES", "地址或者数据库为空"))
  127. }
  128. if config.Conf.DB.Es.IndexB == "" {
  129. log.Error("InitEs", zap.String("IndexB", "indexb bidding 索引为空,请检查"))
  130. } else {
  131. log.Debug("InitEs", zap.String("IndexB", config.Conf.DB.Es.IndexB))
  132. }
  133. if config.Conf.DB.Es.IndexP == "" {
  134. log.Error("InitEs", zap.String("IndexB", "projectset 项目索引为空,请检查"))
  135. } else {
  136. log.Debug("InitEs", zap.String("IndexP", config.Conf.DB.Es.IndexP))
  137. }
  138. if config.Conf.DB.Es.IndexTmp == "" {
  139. log.Error("InitEs", zap.String("IndexTmp 为空", "请检查是否需要配置;该配置主要生产环境需要"))
  140. }
  141. if config.Conf.DB.Es.IndexWinner == "" {
  142. log.Error("InitEs", zap.String("IndexWinner", "中标单位 索引为空,请检查"))
  143. }
  144. if config.Conf.DB.Es.IndexBuyer == "" {
  145. log.Error("InitEs", zap.String("IndexBuyer", "采购单位 索引为空,请检查"))
  146. }
  147. Es1 = &elastic.Elastic{
  148. S_esurl: config.Conf.DB.Es.AddrP,
  149. I_size: config.Conf.DB.Es.Size,
  150. Username: config.Conf.DB.Es.Username,
  151. Password: config.Conf.DB.Es.Password,
  152. }
  153. Es1.InitElasticSize()
  154. if config.Conf.DB.Es.Addr2 != "" {
  155. Es2 = &elastic.Elastic{
  156. S_esurl: config.Conf.DB.Es.Addr2,
  157. I_size: config.Conf.DB.Es.Size,
  158. Username: config.Conf.DB.Es.Username2,
  159. Password: config.Conf.DB.Es.Password2,
  160. }
  161. Es2.InitElasticSize()
  162. fmt.Println("es2", Es2)
  163. }
  164. log.Info("InitEs", zap.Any("duration", time.Since(now).Seconds()))
  165. }
  166. func InitField() {
  167. now := time.Now()
  168. info, _ := MgoB.Find("bidding_processing_field", `{"stype": "project"}`, nil, nil, false, -1, -1)
  169. if len(*info) > 0 {
  170. for _, m := range *info {
  171. if util.IntAll(m["level"]) == 1 {
  172. ProjectField[util.ObjToString(m["field"])] = util.ObjToString(m["ftype"])
  173. } else if util.IntAll(m["level"]) == 2 {
  174. ProjectListF[util.ObjToString(m["field"])] = util.ObjToString(m["ftype"])
  175. }
  176. }
  177. }
  178. log.Info("InitField", zap.Int("ProjectField", len(ProjectField)), zap.Int("ProjectListF", len(ProjectListF)))
  179. log.Info("InitField", zap.Any("duration", time.Since(now).Seconds()))
  180. }
  181. //InitEsBiddingField 初始化 bidding 索引字段
  182. func InitEsBiddingField() {
  183. now := time.Now()
  184. info, _ := MgoB.Find("bidding_processing_field", `{"stype": "bidding"}`, nil, nil, false, -1, -1)
  185. if len(*info) > 0 {
  186. for _, m := range *info {
  187. if util.IntAll(m["level"]) == 1 {
  188. BiddingField[util.ObjToString(m["field"])] = util.ObjToString(m["ftype"])
  189. } else if util.IntAll(m["level"]) == 2 {
  190. pfield := util.ObjToString(m["pfield"])
  191. pfieldMap := BiddingLevelField[pfield]
  192. if pfieldMap == nil {
  193. pfieldMap = make(map[string]string, 0)
  194. }
  195. pfieldMap[util.ObjToString(m["field"])] = util.ObjToString(m["ftype"])
  196. BiddingLevelField[pfield] = pfieldMap
  197. }
  198. }
  199. }
  200. log.Info("InitEsBiddingField", zap.Int("BiddingField es 一级字段数量", len(BiddingField)))
  201. log.Info("InitEsBiddingField", zap.Int("BiddingLevelField es 二级字段数量", len(BiddingLevelField)))
  202. log.Info("InitEsBiddingField", zap.Any("duration", time.Since(now).Seconds()))
  203. }
  204. //verifyESFields 验证es 定义字段类型和 MongoDB 数据字段
  205. func verifyESFields() {
  206. now := time.Now()
  207. log.Info("verifyESFields", zap.String("开始类型检测", ""))
  208. client, _ := es7.NewClient(
  209. es7.SetURL(config.Conf.DB.Es.Addr),
  210. es7.SetBasicAuth(config.Conf.DB.Es.Username, config.Conf.DB.Es.Password),
  211. es7.SetSniff(false),
  212. )
  213. index := config.Conf.DB.Es.IndexB //索引表 bidding
  214. // 获取 Elasticsearch 索引的 mapping 信息
  215. mapping, err := client.GetMapping().Index(index).Do(context.Background())
  216. if err != nil {
  217. log.Error("verifyESFields", zap.Any("getting Elasticsearch mapping:", err))
  218. }
  219. indexName, _ := GetIndexName(client, index)
  220. if indexName == "" || mapping == nil {
  221. log.Error("verifyESFields", zap.String("索引不存在,请检查索引", index))
  222. os.Exit(-1)
  223. return
  224. }
  225. if mapping[indexName].(map[string]interface{})["mappings"] == nil || mapping[indexName].(map[string]interface{})["mappings"].(map[string]interface{})["properties"] == nil {
  226. log.Error("verifyESFields", zap.String("索引不存在或状态不对,请检查索引", index))
  227. os.Exit(-1)
  228. return
  229. }
  230. properties := mapping[indexName].(map[string]interface{})["mappings"].(map[string]interface{})["properties"].(map[string]interface{})
  231. var errField = make([]string, 0)
  232. var okField = make([]string, 0)
  233. var analyzerMap = make(map[string]string) // 分词信息
  234. var esMap = make(map[string]string) //存储es 字段类型
  235. //
  236. for field, ftype := range BiddingField {
  237. eftypeMap, _ := properties[field].(map[string]interface{})
  238. var etype string
  239. var analyzer string
  240. if fftype, ok := eftypeMap["type"]; ok {
  241. etype = fftype.(string)
  242. esMap[field] = etype
  243. }
  244. if ffanalyzer, ok := eftypeMap["analyzer"]; ok {
  245. analyzer = ffanalyzer.(string)
  246. analyzerMap[field] = analyzer
  247. }
  248. if ftype != "" {
  249. if chargeType(ftype, etype) {
  250. okField = append(okField, field)
  251. } else {
  252. errField = append(errField, field)
  253. }
  254. } else {
  255. if field == "_id" {
  256. continue
  257. } else if field == "purchasinglist" || field == "package" || field == "winnerorder" || field == "procurementlist" {
  258. if eproperties, ok := eftypeMap["properties"]; ok {
  259. if eproMap, ok := eproperties.(map[string]interface{}); ok {
  260. for k, v := range eproMap {
  261. if innerMap, ok := v.(map[string]interface{}); ok {
  262. if innerType, ok := innerMap["type"]; ok {
  263. innerLevel := BiddingLevelField[field]
  264. esMap[fmt.Sprintf("%s.%s", field, k)] = innerType.(string)
  265. if chargeType(innerLevel[k], innerType.(string)) {
  266. okField = append(okField, fmt.Sprintf("%s.%s", field, k))
  267. } else {
  268. errField = append(errField, fmt.Sprintf("%s.%s", field, k))
  269. }
  270. }
  271. }
  272. }
  273. }
  274. }
  275. }
  276. }
  277. }
  278. if len(errField) > 0 {
  279. log.Error("verifyESFields", zap.Int("错误字段数量", len(errField)))
  280. for _, field := range errField {
  281. if strings.Contains(field, ".") {
  282. fe := strings.Split(field, ".")
  283. log.Error(fmt.Sprintf("%s 字段类型错误", field), zap.String(fmt.Sprintf("数据库类型为:%s,但是es字段类型是:", BiddingLevelField[fe[0]][fe[1]]), esMap[field]))
  284. } else {
  285. log.Error(fmt.Sprintf("%s 字段类型错误", field), zap.String(fmt.Sprintf("数据库类型为:%s,但是es字段类型是:", BiddingField[field]), esMap[field]))
  286. }
  287. }
  288. os.Exit(-1)
  289. } else {
  290. log.Info("es 字段类型检测结束,", zap.Int("所有字段都符合,检测字段数量为:", len(okField)))
  291. }
  292. log.Info("verifyESFields", zap.Any("duration", time.Since(now).Seconds()))
  293. }
  294. func GetIndexName(client *es7.Client, name string) (string, error) {
  295. // 判断 name 是否为一个别名
  296. res, err := client.Aliases().Alias(name).Do(context.Background())
  297. if err != nil {
  298. // 错误处理
  299. if err.(*es7.Error).Status != 404 && err.(*es7.Error).Details != nil {
  300. return "", err
  301. }
  302. }
  303. if res != nil {
  304. for k, v := range res.Indices {
  305. for _, vv := range v.Aliases {
  306. if vv.AliasName == name {
  307. return k, nil
  308. }
  309. }
  310. }
  311. }
  312. // 判断 name 是否为一个正式索引名称
  313. resa, err := client.IndexExists(name).Do(context.Background())
  314. if err != nil {
  315. // 错误处理
  316. return "", err
  317. }
  318. if resa {
  319. return name, nil
  320. }
  321. // 如果 name 既不是别名,也不是正式索引名称,则返回空字符串
  322. return "", nil
  323. }