elasticutil.go 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158
  1. package elastic
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "log"
  7. "reflect"
  8. "runtime"
  9. "strconv"
  10. "strings"
  11. "sync"
  12. "time"
  13. es "github.com/olivere/elastic/v7"
  14. )
  15. //检索库服务地址
  16. var (
  17. addrs []string
  18. LocCity = map[string]string{}
  19. SIZE = 30
  20. username string
  21. password string
  22. )
  23. const (
  24. QStr = `{"query":{"bool":{"must":[$and],"must_not":[],
  25. "should":[$or],"minimum_should_match" : 1}}}`
  26. )
  27. var pool chan *es.Client
  28. var ntimeout int
  29. var syncPool sync.Pool
  30. //初始化全文检索
  31. func InitElastic(addr string) {
  32. InitElasticSize(addr, SIZE)
  33. }
  34. //自定义HttpClient
  35. /**
  36. var httpclient = &http.Client{Transport: &http.Transport{
  37. Dial: func(netw, addr string) (net.Conn, error) {
  38. deadline := time.Now().Add(5000 * time.Millisecond)
  39. c, err := net.DialTimeout(netw, addr, 10000*time.Millisecond)
  40. if err != nil {
  41. return nil, err
  42. }
  43. tcp_conn := c.(*net.TCPConn)
  44. tcp_conn.SetKeepAlive(false)
  45. tcp_conn.SetDeadline(deadline)
  46. return tcp_conn, nil
  47. },
  48. DisableKeepAlives: true, //不保持,这样才能释放
  49. }}
  50. **/
  51. //var op = es.SetHttpClient(httpclient)
  52. var poolsize = int32(20)
  53. //n倍的池
  54. func InitElasticSize(addr string, size int) {
  55. InitElasticSizeByAuth(addr, size, "", "")
  56. }
  57. //初始化es,带有用户名密码认证
  58. func InitElasticSizeByAuth(addr string, size int, u, p string) {
  59. poolsize = int32(3 * size)
  60. pool = make(chan *es.Client, poolsize)
  61. for _, s := range strings.Split(addr, ",") {
  62. addrs = append(addrs, s)
  63. }
  64. username = u
  65. password = p
  66. for i := 0; i < size; i++ {
  67. client, _ := newClient()
  68. pool <- client
  69. }
  70. }
  71. func newClient() (*es.Client, error) {
  72. opt := []es.ClientOptionFunc{es.SetURL(addrs...), es.SetMaxRetries(2), es.SetSniff(false)}
  73. if username != "" && password != "" {
  74. opt = append(opt, es.SetBasicAuth(username, password))
  75. }
  76. return es.NewClient(opt...)
  77. }
  78. //关闭连接
  79. func DestoryEsConn(client *es.Client) {
  80. select {
  81. case pool <- client:
  82. break
  83. case <-time.After(time.Second * 1):
  84. if client != nil {
  85. client.Stop()
  86. }
  87. client = nil
  88. }
  89. }
  90. var (
  91. lastTime = int64(0)
  92. lastTimeLock = &sync.Mutex{}
  93. )
  94. //获取连接
  95. func GetEsConn() *es.Client {
  96. select {
  97. case c := <-pool:
  98. if c == nil || !c.IsRunning() {
  99. log.Println("new esclient.", len(pool))
  100. client, err := newClient()
  101. if err == nil && client.IsRunning() {
  102. return client
  103. }
  104. }
  105. return c
  106. case <-time.After(time.Second * 4):
  107. //超时
  108. ntimeout++
  109. lastTimeLock.Lock()
  110. defer lastTimeLock.Unlock()
  111. //12秒后允许创建链接
  112. c := time.Now().Unix() - lastTime
  113. if c > 12 {
  114. lastTime = time.Now().Unix()
  115. log.Println("add client..", len(pool))
  116. c, _ := newClient()
  117. go func() {
  118. for i := 0; i < 2; i++ {
  119. client, _ := newClient()
  120. pool <- client
  121. }
  122. }()
  123. return c
  124. }
  125. return nil
  126. }
  127. }
  128. //保存对象
  129. func Save(index, itype string, obj interface{}) bool {
  130. client := GetEsConn()
  131. defer DestoryEsConn(client)
  132. defer func() {
  133. if r := recover(); r != nil {
  134. log.Println("[E]", r)
  135. for skip := 1; ; skip++ {
  136. _, file, line, ok := runtime.Caller(skip)
  137. if !ok {
  138. break
  139. }
  140. go log.Printf("%v,%v\n", file, line)
  141. }
  142. }
  143. }()
  144. data := objToMap(obj)
  145. _id := bsonIdToSId((*data)["_id"])
  146. (*data)["id"] = _id
  147. delete((*data), "_id")
  148. _, err := client.Index().Index(index).Id(_id).BodyJson(data).Do(context.TODO())
  149. if err != nil {
  150. log.Println("保存到ES出错", err.Error(), obj)
  151. return false
  152. } else {
  153. return true
  154. }
  155. }
  156. //通用查询
  157. //{"query": {"bool":{"must":[{"query_string":{"default_field":"name","query":"*"}}]}}}
  158. //{"query":{"bool":{"must":{"match":{"content":{"query":"fulltextsearch","operator":"and"}}},"should":[{"match":{"content":{"query":"Elasticsearch","boost":3}}},{"match":{"content":{"query":"Lucene","boost":2}}}]}}}
  159. //prefix
  160. //{"query":{"match":{"title":{"query":"brownfox","operator":"and"}}}} //默认为or
  161. //{"query":{"multi_match":{"query":"PolandStreetW1V","type":"most_fields","fields":["*_street","city^2","country","postcode"]}}}
  162. //{"query":{"wildcard":{"postcode":"W?F*HW"}}}
  163. //{"query":{"regexp":{"postcode":"W[0-9].+"}}}
  164. //{"query":{"filtered":{"filter":{"range":{"price":{"gte":10000}}}}},"aggs":{"single_avg_price":{"avg":{"field":"price"}}}}
  165. //{"query":{"match":{"make":"ford"}},"aggs":{"colors":{"terms":{"field":"color"}}}}//查fork有几种颜色
  166. //过滤器不会计算相关度的得分,所以它们在计算上更快一些
  167. //{"query":{"filtered":{"query":{"match_all":{}},"filter":{"range":{"balance":{"gte":20000,"lte":30000}}}}}}
  168. //{"query":{"match_all":{}},"from":10,"size":10,"_source":["account_number","balance"],"sort":{"balance":{"order":"desc"}}}
  169. //{"query":{"match_phrase":{"address":"milllane"}}}和match不同会去匹配整个短语,相当于must[]
  170. func GetBySearchType(index, searchType, query string) (int64, *[]map[string]interface{}) {
  171. t, _, l := get(index, "", searchType, query, true, true)
  172. return t, l
  173. }
  174. func Get(index, itype, query string) *[]map[string]interface{} {
  175. _, _, r := get(index, itype, query, "", true, true)
  176. return r
  177. }
  178. func get(index, itype, query, searchType string, isLimit, isHighlight bool) (int64, int, *[]map[string]interface{}) {
  179. //log.Println("query -- ", query)
  180. client := GetEsConn()
  181. defer func() {
  182. go DestoryEsConn(client)
  183. }()
  184. var res []map[string]interface{}
  185. var total int64
  186. var resNum int
  187. if client != nil {
  188. defer func() {
  189. if r := recover(); r != nil {
  190. log.Println("[E]", r)
  191. for skip := 1; ; skip++ {
  192. _, file, line, ok := runtime.Caller(skip)
  193. if !ok {
  194. break
  195. }
  196. go log.Printf("%v,%v\n", file, line)
  197. }
  198. }
  199. }()
  200. ss := client.Search().Index(index).Source(query)
  201. if searchType != "" {
  202. ss.SearchType(searchType)
  203. }
  204. searchResult, err := ss.Do(context.TODO())
  205. if err != nil {
  206. log.Println("从ES查询出错", err.Error())
  207. return total, resNum, nil
  208. }
  209. total = searchResult.TotalHits()
  210. if searchResult.Hits != nil {
  211. resNum = len(searchResult.Hits.Hits)
  212. if isLimit && resNum < 5000 {
  213. res = make([]map[string]interface{}, resNum)
  214. for i, hit := range searchResult.Hits.Hits {
  215. //d := json.NewDecoder(bytes.NewBuffer(*hit.Source))
  216. //d.UseNumber()
  217. //d.Decode(&res[i])
  218. parseErr := json.Unmarshal(hit.Source, &res[i])
  219. if res[i] != nil {
  220. res[i]["_id"] = hit.Id
  221. }
  222. if isHighlight && parseErr == nil && hit.Highlight != nil && res[i] != nil {
  223. res[i]["highlight"] = map[string][]string(hit.Highlight)
  224. }
  225. }
  226. } else {
  227. log.Println("查询结果太多,查询到:", resNum, "条")
  228. }
  229. }
  230. }
  231. return total, resNum, &res
  232. }
  233. func GetOA(index, itype, query string) (*[]map[string]interface{}, int) {
  234. _, n, l := get(index, itype, query, "", true, true)
  235. return l, n
  236. }
  237. func GetNoLimit(index, itype, query string) *[]map[string]interface{} {
  238. _, _, l := get(index, itype, query, "", false, false)
  239. return l
  240. }
  241. //分页查询
  242. //{"name":"张三","$and":[{"age":{"$gt":10}},{"age":{"$lte":20}}]}
  243. //fields直接是 `"_id","title"`
  244. func GetPage(index, itype, query, order, field string, start, limit int) *[]map[string]interface{} {
  245. return Get(index, itype, MakeQuery(query, order, field, start, limit))
  246. }
  247. //openapi
  248. func GetOAPage(index, itype, query, order, field string, start, limit int) (*[]map[string]interface{}, int) {
  249. return GetOA(index, itype, MakeQuery(query, order, field, start, limit))
  250. }
  251. var SR = strings.Replace
  252. func MakeQuery(query, order, fileds string, start, limit int) string {
  253. res := AnalyQuery(query, "", QStr)
  254. if len(res) > 10 {
  255. res = SR(SR(SR(SR(res, ",$and", "", -1), "$and", "", -1), ",$or", "", -1), "$or", "", -1)
  256. if len(fileds) > 0 {
  257. //"_source":["account_number","balance"]
  258. res = res[:len(res)-1] + `,"_source":[` + fileds + "]}"
  259. }
  260. //{"name":-1,"age":1}
  261. if len(order) > 0 {
  262. res = res[:len(res)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  263. }
  264. if start > -1 {
  265. res = res[:len(res)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  266. }
  267. return res
  268. }
  269. return ""
  270. }
  271. //{"name":"aaa"}
  272. func AnalyQuery(query interface{}, parent string, result string) string {
  273. m := make(map[string]interface{})
  274. if q1, ok := query.(string); ok {
  275. json.Unmarshal([]byte(q1), &m)
  276. } else if q2, ok2 := query.(map[string]interface{}); ok2 {
  277. m = q2
  278. }
  279. if len(parent) == 0 {
  280. for k, v := range m {
  281. if k == "$and" || k == "$or" {
  282. temps := ""
  283. if map1, ok := v.([]interface{}); ok {
  284. for i := 0; i < len(map1); i++ {
  285. temps += "," + AnalyQuery(map1[i], k, "")
  286. }
  287. }
  288. if len(temps) > 0 {
  289. temps = temps[1:]
  290. }
  291. result = SR(result, k, temps+","+k, 1)
  292. } else {
  293. switch reflect.TypeOf(v).String() {
  294. case "string":
  295. if strings.Index(k, "TERM_") == 0 {
  296. result = SR(result, "$and", `{"term":{"`+SR(k, "TERM_", "", 1)+`":"`+fmt.Sprintf("%v", v)+`"}},$and`, 1)
  297. } else {
  298. result = SR(result, "$and", `{"query_string":{"default_field":"`+k+`","query":"`+fmt.Sprintf("%v", v)+`"}},$and`, 1)
  299. }
  300. case "int", "int8", "int32", "int64", "float32", "float64":
  301. if strings.Index(k, "TERM_") == 0 {
  302. result = SR(result, "$and", `{"term":{"`+SR(k, "TERM_", "", 1)+`":`+fmt.Sprintf("%v", v)+`}},$and`, 1)
  303. } else {
  304. result = SR(result, "$and", `{"query_string":{"default_field":"`+k+`","query":`+fmt.Sprintf("%v", v)+`}},$and`, 1)
  305. }
  306. default:
  307. result = SR(result, "$and", AnalyQuery(v, k, "")+",$and", 1)
  308. }
  309. }
  310. }
  311. return result
  312. } else {
  313. for k, v := range m {
  314. if k == "$in" {
  315. s := ""
  316. if map1, ok := v.([]interface{}); ok {
  317. for i := 0; i < len(map1); i++ {
  318. s += "," + `"` + fmt.Sprintf("%v", map1[i]) + `"`
  319. }
  320. }
  321. if len(s) > 0 {
  322. s = s[1:]
  323. }
  324. return `{"terms":{"` + parent + `":[` + s + `]}}`
  325. } else if strings.Contains(k, "$lt") || strings.Contains(k, "$gt") {
  326. return `{"range":{"` + parent + `":{"` + SR(k, "$", "", 1) + `":` + fmt.Sprintf("%v", v) + `}}}`
  327. } else {
  328. switch reflect.TypeOf(v).String() {
  329. case "string":
  330. if strings.Index(k, "TERM_") == 0 {
  331. return `{"term":{"` + SR(k, "TERM_", "", 1) + `":"` + fmt.Sprintf("%v", v) + `"}}`
  332. } else {
  333. return `{"query_string":{"default_field":"` + k + `","query":"` + fmt.Sprintf("%v", v) + `"}}`
  334. }
  335. case "int", "int8", "int32", "int64", "float32", "float64":
  336. if strings.Index(k, "TERM_") == 0 {
  337. return `{"term":{"` + SR(k, "TERM_", "", 1) + `":` + fmt.Sprintf("%v", v) + `}}`
  338. } else {
  339. return `{"query_string":{"default_field":"` + k + `","query":` + fmt.Sprintf("%v", v) + `}}`
  340. }
  341. default:
  342. return AnalyQuery(v, k, result)
  343. }
  344. }
  345. }
  346. }
  347. return result
  348. }
  349. func GetByIdField(index, itype, id, fields string) *map[string]interface{} {
  350. client := GetEsConn()
  351. defer DestoryEsConn(client)
  352. if client != nil {
  353. defer func() {
  354. if r := recover(); r != nil {
  355. log.Println("[E]", r)
  356. for skip := 1; ; skip++ {
  357. _, file, line, ok := runtime.Caller(skip)
  358. if !ok {
  359. break
  360. }
  361. go log.Printf("%v,%v\n", file, line)
  362. }
  363. }
  364. }()
  365. query := `{"query":{"term":{"_id":"` + id + `"}}`
  366. if len(fields) > 0 {
  367. query = query + `,"_source":[` + fields + `]`
  368. }
  369. query = query + "}"
  370. searchResult, err := client.Search().Index(index).Source(query).Do(context.TODO())
  371. if err != nil {
  372. log.Println("从ES查询出错", err.Error())
  373. return nil
  374. }
  375. var res map[string]interface{}
  376. if searchResult.Hits != nil {
  377. resNum := len(searchResult.Hits.Hits)
  378. if resNum == 1 {
  379. res = make(map[string]interface{})
  380. for _, hit := range searchResult.Hits.Hits {
  381. json.Unmarshal(hit.Source, &res)
  382. if res != nil {
  383. res["_id"] = hit.Id
  384. }
  385. }
  386. return &res
  387. }
  388. }
  389. }
  390. return nil
  391. }
  392. //根据id来查询文档
  393. func GetById(index, itype string, ids ...string) *[]map[string]interface{} {
  394. client := GetEsConn()
  395. defer DestoryEsConn(client)
  396. var res []map[string]interface{}
  397. if client != nil {
  398. defer func() {
  399. if r := recover(); r != nil {
  400. log.Println("[E]", r)
  401. for skip := 1; ; skip++ {
  402. _, file, line, ok := runtime.Caller(skip)
  403. if !ok {
  404. break
  405. }
  406. go log.Printf("%v,%v\n", file, line)
  407. }
  408. }
  409. }()
  410. query := es.NewIdsQuery().Ids(ids...)
  411. searchResult, err := client.Search().Index(index).Query(query).Do(context.TODO())
  412. if err != nil {
  413. log.Println("从ES查询出错", err.Error())
  414. return nil
  415. }
  416. if searchResult.Hits != nil {
  417. resNum := len(searchResult.Hits.Hits)
  418. if resNum < 5000 {
  419. res = make([]map[string]interface{}, resNum)
  420. for i, hit := range searchResult.Hits.Hits {
  421. json.Unmarshal(hit.Source, &res[i])
  422. if res[i] != nil {
  423. res[i]["_id"] = hit.Id
  424. }
  425. }
  426. } else {
  427. log.Println("查询结果太多,查询到:", resNum, "条")
  428. }
  429. }
  430. }
  431. return &res
  432. }
  433. //根据语句更新对象
  434. func Update(index, itype, id string, updateStr string) bool {
  435. client := GetEsConn()
  436. defer DestoryEsConn(client)
  437. b := false
  438. if client != nil {
  439. defer func() {
  440. if r := recover(); r != nil {
  441. log.Println("[E]", r)
  442. for skip := 1; ; skip++ {
  443. _, file, line, ok := runtime.Caller(skip)
  444. if !ok {
  445. break
  446. }
  447. go log.Printf("%v,%v\n", file, line)
  448. }
  449. }
  450. }()
  451. var err error
  452. esc := es.NewScript(updateStr)
  453. esc.Lang("groovy")
  454. _, err = client.Update().Index(index).Type(itype).Id(id).Script(esc).Do(context.TODO())
  455. if err != nil {
  456. log.Println("更新检索出错:", err.Error())
  457. } else {
  458. b = true
  459. }
  460. }
  461. return b
  462. }
  463. func BulkUpdate(index, itype string, ids []string, updateStr string) {
  464. client := GetEsConn()
  465. defer DestoryEsConn(client)
  466. if client != nil {
  467. defer func() {
  468. if r := recover(); r != nil {
  469. log.Println("[E]", r)
  470. for skip := 1; ; skip++ {
  471. _, file, line, ok := runtime.Caller(skip)
  472. if !ok {
  473. break
  474. }
  475. go log.Printf("%v,%v\n", file, line)
  476. }
  477. }
  478. }()
  479. for _, id := range ids {
  480. esc := es.NewScript(updateStr)
  481. esc.Lang("groovy")
  482. _, err := client.Update().Index(index).Type(itype).Id(id).Script(esc).Do(context.TODO())
  483. if err != nil {
  484. log.Println("更新检索出错:", err.Error())
  485. }
  486. }
  487. }
  488. }
  489. //根据id删除索引对象
  490. func DelById(index, itype, id string) bool {
  491. client := GetEsConn()
  492. defer DestoryEsConn(client)
  493. b := false
  494. if client != nil {
  495. defer func() {
  496. if r := recover(); r != nil {
  497. log.Println("[E]", r)
  498. for skip := 1; ; skip++ {
  499. _, file, line, ok := runtime.Caller(skip)
  500. if !ok {
  501. break
  502. }
  503. go log.Printf("%v,%v\n", file, line)
  504. }
  505. }
  506. }()
  507. var err error
  508. _, err = client.Delete().Index(index).Id(id).Do(context.TODO())
  509. if err != nil {
  510. log.Println("更新检索出错:", err.Error())
  511. } else {
  512. b = true
  513. }
  514. }
  515. return b
  516. }
  517. //先删除后增
  518. func UpdateNewDoc(index, itype string, obj ...interface{}) bool {
  519. client := GetEsConn()
  520. defer DestoryEsConn(client)
  521. b := false
  522. if client != nil {
  523. defer func() {
  524. if r := recover(); r != nil {
  525. log.Println("[E]", r)
  526. for skip := 1; ; skip++ {
  527. _, file, line, ok := runtime.Caller(skip)
  528. if !ok {
  529. break
  530. }
  531. go log.Printf("%v,%v\n", file, line)
  532. }
  533. }
  534. }()
  535. var err error
  536. for _, v := range obj {
  537. tempObj := objToMap(v)
  538. if tempObj == nil || len(*tempObj) == 0 {
  539. continue
  540. }
  541. id := bsonIdToSId((*tempObj)["_id"])
  542. (*tempObj)["id"] = id
  543. delete(*tempObj, "_id")
  544. if id != "" {
  545. client.Delete().Index(index).Id(id).Do(context.TODO())
  546. }
  547. _, err = client.Index().Index(index).Id(id).BodyJson(tempObj).Do(context.TODO())
  548. if err != nil {
  549. log.Println("保存到ES出错", err.Error())
  550. } else {
  551. b = true
  552. }
  553. }
  554. }
  555. return b
  556. }
  557. func BulkSave(index, itype string, obj *[]map[string]interface{}, isDelBefore bool) {
  558. client := GetEsConn()
  559. defer DestoryEsConn(client)
  560. if client != nil {
  561. defer func() {
  562. if r := recover(); r != nil {
  563. log.Println("[E]", r)
  564. for skip := 1; ; skip++ {
  565. _, file, line, ok := runtime.Caller(skip)
  566. if !ok {
  567. break
  568. }
  569. go log.Printf("%v,%v\n", file, line)
  570. }
  571. }
  572. }()
  573. req := client.Bulk()
  574. for _, v := range *obj {
  575. if v == nil || len(v) == 0 {
  576. continue
  577. }
  578. _id := bsonIdToSId(v["_id"])
  579. v["id"] = _id
  580. delete(v, "_id")
  581. if isDelBefore && _id != "" {
  582. req = req.Add(es.NewBulkDeleteRequest().Index(index).Id(_id))
  583. }
  584. req = req.Add(es.NewBulkIndexRequest().Index(index).Id(_id).Doc(v))
  585. }
  586. _, err := req.Do(context.TODO())
  587. if err != nil {
  588. log.Println("批量保存到ES出错", err.Error())
  589. }
  590. }
  591. }
  592. func Count(index, itype string, query interface{}) int64 {
  593. client := GetEsConn()
  594. defer DestoryEsConn(client)
  595. if client != nil {
  596. defer func() {
  597. if r := recover(); r != nil {
  598. log.Println("[E]", r)
  599. for skip := 1; ; skip++ {
  600. _, file, line, ok := runtime.Caller(skip)
  601. if !ok {
  602. break
  603. }
  604. go log.Printf("%v,%v\n", file, line)
  605. }
  606. }
  607. }()
  608. var n int64
  609. var err error
  610. if qs, ok := query.(string); ok {
  611. n, err = client.Count(index).BodyString(qs).Do(context.TODO())
  612. } else if qi, ok2 := query.(es.Query); ok2 {
  613. n, err = client.Count(index).Query(qi).Do(context.TODO())
  614. }
  615. if err != nil {
  616. log.Println("统计出错", err.Error())
  617. }
  618. return n
  619. }
  620. return 0
  621. }
  622. //ngram精确查询
  623. /*
  624. {
  625. "query": {
  626. "bool": {
  627. "should": [
  628. {
  629. "bool":{
  630. "must":[
  631. { "multi_match": {
  632. "query": "智能",
  633. "type": "phrase",
  634. "fields": [
  635. "title"
  636. ],
  637. "analyzer": "my_ngram"
  638. }
  639. },{
  640. "multi_match": {
  641. "query": "机器",
  642. "type": "phrase",
  643. "fields": [
  644. "title"
  645. ],
  646. "analyzer": "my_ngram"
  647. }
  648. },{
  649. "multi_match": {
  650. "query": "2016",
  651. "type": "phrase",
  652. "fields": [
  653. "title"
  654. ],
  655. "analyzer": "my_ngram"
  656. }
  657. }
  658. ]
  659. }
  660. },
  661. {
  662. "bool":{
  663. "must":[
  664. { "multi_match": {
  665. "query": "河南",
  666. "type": "phrase",
  667. "fields": [
  668. "title"
  669. ],
  670. "analyzer": "my_ngram"
  671. }
  672. },{
  673. "multi_match": {
  674. "query": "工商",
  675. "type": "phrase",
  676. "fields": [
  677. "title"
  678. ],
  679. "analyzer": "my_ngram"
  680. }
  681. },{
  682. "multi_match": {
  683. "query": "2016",
  684. "type": "phrase",
  685. "fields": [
  686. "title"
  687. ],
  688. "analyzer": "my_ngram"
  689. }
  690. }
  691. ]
  692. }
  693. }
  694. ],"minimum_should_match": 1
  695. }
  696. },
  697. "_source": [
  698. "_id",
  699. "title"
  700. ],
  701. "from": 0,
  702. "size": 10,
  703. "sort": [{
  704. "publishtime": "desc"
  705. }]
  706. }
  707. */
  708. //"2016+智能+办公,"河南+工商"
  709. //["2016+智能+办公","河南+工商"]
  710. //QStr = `{"query":{"bool":{should":[$or],"minimum_should_match" : 1}}}`
  711. //{"bool":{"must":[]}}
  712. //{"multi_match": {"query": "$word","type": "phrase", "fields": [$field],"analyzer": "my_ngram"}}
  713. //"highlight": {"pre_tags": [""],"post_tags": [""],"fields": {"detail": {"fragment_size": 1,"number_of_fragments": 1},"title": {"fragment_size": 1,"number_of_fragments": 1}}}
  714. const (
  715. //此处最后少一个},正好NgramStr取[1:]多一个}
  716. FilterQuery = `{"query": {"filtered": {"filter": {"bool": {"must": [%s]}},%s}}`
  717. NgramStr = `{"query":{"bool":{"must":[%s],"should":[%s],"minimum_should_match": 1}}}`
  718. NgramMust = `{"bool":{"must":[%s]}}`
  719. NgramMustAndNot = `{"bool":{"must":[%s],"must_not":[%s]}}`
  720. minq = `{"multi_match": {"query": "%s","type": "phrase", "fields": [%s]}}`
  721. HL = `"highlight": {"pre_tags": [""],"post_tags": [""],"fields": {%s}}`
  722. highlightStr = `%s: {"fragment_size": %d,"number_of_fragments": 1}`
  723. FilterQuery_New = `{"query":{"bool":{"must": [%s%s%s],"should":[]}}}`
  724. MatchQueryString = `{"match": {%s: { "query":"%s", "operator": "and"}}}`
  725. HL_New = `"highlight": {"pre_tags": ["<HL>"],"post_tags": ["<HL>"],"fields": {%s}}`
  726. //数据查询高亮标记2019-07-10
  727. HL_MP = `"highlight": {"pre_tags": ["<HL>"],"post_tags": ["</HL>"],"fields": {%s}}`
  728. ik_highlightStr = `%s: {"fragment_size": %d,"number_of_fragments": 1,"require_field_match": true}`
  729. IK_pre_tags = `<font class=\"es-highlight\">`
  730. IK_post_tags = `</font>`
  731. HL_IK = `"highlight": {"pre_tags": ["` + IK_pre_tags + `"],"post_tags": ["` + IK_post_tags + `"],"fields": {%s}}`
  732. )
  733. //替换了"号
  734. func GetNgramQuery(query interface{}, mustquery, findfields string) (qstr string) {
  735. var words []string
  736. if q, ok := query.(string); ok {
  737. if q != "" {
  738. words = strings.Split(q, ",")
  739. }
  740. } else if q, ok := query.([]string); ok {
  741. words = q
  742. } else if q, ok := query.([]interface{}); ok {
  743. words = objArrToStringArr(q)
  744. }
  745. if words != nil {
  746. new_minq := fmt.Sprintf(minq, "%s", findfields)
  747. musts := []string{}
  748. for _, qs_words := range words {
  749. qws := strings.Split(qs_words, "+")
  750. mq := []string{}
  751. for _, qs_word := range qws {
  752. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
  753. }
  754. musts = append(musts, fmt.Sprintf(NgramMust, strings.Join(mq, ",")))
  755. }
  756. qstr = fmt.Sprintf(NgramStr, mustquery, strings.Join(musts, ","))
  757. //log.Println("ngram-query", qstr)
  758. } else {
  759. qstr = fmt.Sprintf(NgramStr, mustquery, "")
  760. }
  761. return
  762. }
  763. func GetNgramQuery_New(querystring, querymust interface{}, must, findfields string) (qstring string) {
  764. querymust_string := ""
  765. var wordsMust []string
  766. if q, ok := querymust.(string); ok {
  767. if q != "" {
  768. wordsMust = strings.Split(q, ",")
  769. }
  770. } else if q, ok := querymust.([]string); ok {
  771. wordsMust = q
  772. } else if q, ok := querymust.([]interface{}); ok {
  773. wordsMust = objArrToStringArr(q)
  774. }
  775. if wordsMust != nil {
  776. new_minq := fmt.Sprintf(minq, "%s", findfields)
  777. musts := []string{}
  778. for _, qs_wordsMust := range wordsMust {
  779. qws := strings.Split(qs_wordsMust, "+")
  780. mq := []string{}
  781. for _, qs_word := range qws {
  782. mq = append(mq, fmt.Sprintf(new_minq, qs_word))
  783. }
  784. musts = append(musts, fmt.Sprintf(NgramMust, strings.Join(mq, ",")))
  785. }
  786. querymust_string = strings.Join(musts, ",")
  787. }
  788. //log.Println("must", must, querymust_string)
  789. //querystring---------------------------------------------
  790. query_string := ""
  791. var querysShold []string
  792. if q, ok := querystring.(string); ok {
  793. if q != "" {
  794. querysShold = strings.Split(q, ",")
  795. }
  796. } else if q, ok := querystring.([]string); ok {
  797. querysShold = q
  798. } else if q, ok := querystring.([]interface{}); ok {
  799. querysShold = objArrToStringArr(q)
  800. }
  801. if querysShold != nil {
  802. for k, name := range strings.Split(findfields, ",") {
  803. for _, qs_querysShold := range querysShold {
  804. if k > 0 {
  805. query_string = query_string + "," + fmt.Sprintf(MatchQueryString, fmt.Sprint(name), qs_querysShold)
  806. } else {
  807. query_string = query_string + fmt.Sprintf(MatchQueryString, fmt.Sprint(name), qs_querysShold)
  808. }
  809. }
  810. }
  811. }
  812. //log.Println("querystring", query_string)
  813. if querymust_string == "" {
  814. qstring = fmt.Sprintf(FilterQuery_New, must, query_string, querymust_string)
  815. } else {
  816. qstring = fmt.Sprintf(FilterQuery_New, must, query_string, ","+querymust_string)
  817. }
  818. return
  819. }
  820. func GetByNgram(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int) *[]map[string]interface{} {
  821. return GetByNgramAll(index, itype, query, mustquery, findfields, order, fields, start, limit, false, false)
  822. }
  823. //增加高亮、过滤查询、高亮截取字数
  824. func GetByNgramOther(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool, count int) *[]map[string]interface{} {
  825. defer catch()
  826. qstr := ""
  827. if mustquery != "" && filtermode {
  828. qstr = GetNgramQuery(query, "", findfields)
  829. qstr = fmt.Sprintf(FilterQuery, mustquery, qstr[1:])
  830. } else {
  831. qstr = GetNgramQuery(query, mustquery, findfields)
  832. }
  833. if qstr != "" {
  834. if highlight {
  835. ws := []string{}
  836. for _, w := range strings.Split(findfields, ",") {
  837. ws = append(ws, fmt.Sprintf(highlightStr, w, count))
  838. }
  839. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  840. }
  841. if len(fields) > 0 {
  842. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  843. }
  844. if len(order) > 0 {
  845. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  846. }
  847. if start > -1 {
  848. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  849. }
  850. //log.Println("ngram-find", qstr)
  851. return Get(index, itype, qstr)
  852. } else {
  853. return nil
  854. }
  855. }
  856. //增加高亮、过滤查询
  857. //替换了"号
  858. func GetByNgramAll(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool) *[]map[string]interface{} {
  859. defer catch()
  860. qstr := ""
  861. if mustquery != "" && filtermode {
  862. qstr = GetNgramQuery(query, "", findfields)
  863. qstr = fmt.Sprintf(FilterQuery, mustquery, qstr[1:])
  864. } else {
  865. qstr = GetNgramQuery(query, mustquery, findfields)
  866. }
  867. if qstr != "" {
  868. if highlight {
  869. ws := []string{}
  870. for _, w := range strings.Split(findfields, ",") {
  871. ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
  872. }
  873. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  874. }
  875. if len(fields) > 0 {
  876. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  877. }
  878. if strings.HasPrefix(order, "CUSTOM_") {
  879. qstr = qstr[:len(qstr)-1] + `,` + strings.TrimLeft(order, "CUSTOM_") + `}`
  880. } else if len(order) > 0 {
  881. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  882. }
  883. if start > -1 {
  884. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  885. }
  886. //log.Println("ngram-find", qstr)
  887. return Get(index, itype, qstr)
  888. } else {
  889. return nil
  890. }
  891. }
  892. //增加高亮、过滤查询
  893. func GetByNgramAll_New(index, itype string, querystring, querymust interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool) *[]map[string]interface{} {
  894. defer catch()
  895. qstr := ""
  896. if filtermode {
  897. qstr = GetNgramQuery_New(querystring, querymust, mustquery, findfields)
  898. } else {
  899. qstr = GetNgramQuery_New(querystring, "", mustquery, findfields)
  900. }
  901. if qstr != "" {
  902. if highlight {
  903. ws := []string{}
  904. for _, w := range strings.Split(findfields, ",") {
  905. ws = append(ws, w+`:{"force_source": true}`)
  906. }
  907. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_New, strings.Join(ws, ",")) + `}`
  908. }
  909. if len(fields) > 0 {
  910. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  911. }
  912. if len(order) > 0 {
  913. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", ",", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  914. }
  915. if start > -1 {
  916. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  917. }
  918. //log.Println("ngram-find", order, qstr)
  919. return Get(index, itype, qstr)
  920. } else {
  921. return nil
  922. }
  923. }
  924. type KeyConfig struct {
  925. Keys []string `json:"key"`
  926. NotKeys []string `json:"notkey"`
  927. InfoTypes []string `json:"infotype"`
  928. Areas []string `json:"area"`
  929. }
  930. //替换了"号
  931. func GetResForJY(index, itype string, keys []KeyConfig, allquery, findfields, SortQuery, fields string, start, limit int) *[]map[string]interface{} {
  932. if len(keys) > 0 {
  933. qstr := ""
  934. new_minq := fmt.Sprintf(minq, "%s", findfields)
  935. not_new_minq := fmt.Sprintf(minq, "%s", findfields) //排除词只查询标题
  936. musts := []string{}
  937. for _, qs_words := range keys {
  938. mq := []string{}
  939. notmq := []string{}
  940. for _, qs_word := range qs_words.Keys {
  941. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
  942. /*
  943. qs := AnalyzerWord("bidding", qs_word)
  944. for _, qw := range qs {
  945. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qw)))
  946. }
  947. */
  948. }
  949. for _, qs_word := range qs_words.NotKeys {
  950. notmq = append(notmq, fmt.Sprintf(not_new_minq, ReplaceYH(qs_word)))
  951. }
  952. if len(qs_words.Areas) > 0 {
  953. mq = append(mq, fmt.Sprintf(`{"terms":{"area":["%s"]}}`, strings.Join(qs_words.Areas, `","`)))
  954. }
  955. if len(qs_words.InfoTypes) > 0 {
  956. mq = append(mq, fmt.Sprintf(`{"terms":{"toptype":["%s"]}}`, strings.Join(qs_words.InfoTypes, `","`)))
  957. }
  958. musts = append(musts, fmt.Sprintf(NgramMustAndNot, strings.Join(mq, ","), strings.Join(notmq, ",")))
  959. }
  960. qstr = fmt.Sprintf(NgramStr, "", strings.Join(musts, ","))
  961. qstr = fmt.Sprintf(FilterQuery, allquery, qstr[1:])
  962. ws := []string{}
  963. for _, w := range strings.Split(findfields, ",") {
  964. ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
  965. }
  966. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  967. if len(fields) > 0 {
  968. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  969. }
  970. if len(SortQuery) > 0 {
  971. qstr = qstr[:len(qstr)-1] + `,"sort":` + SortQuery + `}`
  972. }
  973. if start > -1 {
  974. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  975. }
  976. //log.Println("jy-ngram-find", qstr)
  977. return Get(index, itype, qstr)
  978. } else {
  979. return nil
  980. }
  981. }
  982. func ReplaceYH(src string) (rpl string) {
  983. return strings.Replace(src, `"`, `\"`, -1)
  984. }
  985. //
  986. func GetAllByNgram(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
  987. if qstr != "" {
  988. if highlight {
  989. ws := []string{}
  990. for _, w := range strings.Split(findfields, ",") {
  991. ws = append(ws, fmt.Sprintf(highlightStr, w, count))
  992. }
  993. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  994. }
  995. if len(fields) > 0 {
  996. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  997. }
  998. if len(order) > 0 {
  999. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  1000. }
  1001. if start > -1 {
  1002. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1003. }
  1004. log.Println("GetAllByNgram:", qstr)
  1005. return Get(index, itype, qstr)
  1006. } else {
  1007. return nil
  1008. }
  1009. }
  1010. //数据标记2019-07-10
  1011. func GetAllByNgram_MP(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
  1012. if qstr != "" {
  1013. if highlight {
  1014. ws := []string{}
  1015. for _, w := range strings.Split(findfields, ",") {
  1016. ws = append(ws, fmt.Sprintf(highlightStr, w, count))
  1017. }
  1018. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_MP, strings.Join(ws, ",")) + `}`
  1019. }
  1020. if len(fields) > 0 {
  1021. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  1022. }
  1023. if len(order) > 0 {
  1024. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  1025. }
  1026. if start > -1 {
  1027. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1028. }
  1029. // log.Println("GetAllByNgram:", qstr)
  1030. return Get(index, itype, qstr)
  1031. } else {
  1032. return nil
  1033. }
  1034. }
  1035. //ik 分词
  1036. func GetAllByIk(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
  1037. if qstr != "" {
  1038. if highlight {
  1039. ws := []string{}
  1040. for _, w := range strings.Split(findfields, ",") {
  1041. ws = append(ws, fmt.Sprintf(ik_highlightStr, w, count))
  1042. }
  1043. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_IK, strings.Join(ws, ",")) + `}`
  1044. }
  1045. if len(fields) > 0 {
  1046. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  1047. }
  1048. if len(order) > 0 {
  1049. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  1050. }
  1051. if start > -1 {
  1052. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1053. }
  1054. //log.Println("GetAllByNgram:", qstr)
  1055. return Get(index, itype, qstr)
  1056. } else {
  1057. return nil
  1058. }
  1059. }
  1060. func GetResForJYView(index, itype string, keys []KeyConfig, allquery, findfields, SortQuery, fields string, start, limit int) *[]map[string]interface{} {
  1061. if len(keys) > 0 {
  1062. qstr := ""
  1063. new_minq := fmt.Sprintf(minq, "%s", findfields)
  1064. not_new_minq := fmt.Sprintf(minq, "%s", findfields) //排除词只查询标题
  1065. musts := []string{}
  1066. for _, qs_words := range keys {
  1067. mq := []string{}
  1068. notmq := []string{}
  1069. for _, qs_word := range qs_words.Keys {
  1070. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
  1071. }
  1072. for _, qs_word := range qs_words.NotKeys {
  1073. notmq = append(notmq, fmt.Sprintf(not_new_minq, ReplaceYH(qs_word)))
  1074. }
  1075. if len(qs_words.Areas) > 0 {
  1076. mq = append(mq, fmt.Sprintf(`{"terms":{"area":["%s"]}}`, strings.Join(qs_words.Areas, `","`)))
  1077. }
  1078. if len(qs_words.InfoTypes) > 0 {
  1079. mq = append(mq, fmt.Sprintf(`{"terms":{"toptype":["%s"]}}`, strings.Join(qs_words.InfoTypes, `","`)))
  1080. }
  1081. musts = append(musts, fmt.Sprintf(NgramMustAndNot, strings.Join(mq, ","), strings.Join(notmq, ",")))
  1082. }
  1083. qstr = fmt.Sprintf(NgramStr, "", strings.Join(musts, ","))
  1084. qstr = fmt.Sprintf(FilterQuery, allquery, qstr[1:])
  1085. ws := []string{}
  1086. for _, w := range strings.Split(findfields, ",") {
  1087. ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
  1088. }
  1089. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  1090. if len(fields) > 0 {
  1091. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  1092. }
  1093. if len(SortQuery) > 0 {
  1094. qstr = qstr[:len(qstr)-1] + `,"sort":` + SortQuery + `}`
  1095. }
  1096. if start > -1 {
  1097. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1098. }
  1099. return Get(index, itype, qstr)
  1100. } else {
  1101. return nil
  1102. }
  1103. }