elasticutil.go 33 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154
  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 DfsGet(index, query string) (int64, *[]map[string]interface{}) {
  171. t, _, l := get(index, "", query, "dfs_query_then_fetch", 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. searchResult, err := client.Search().Index(index).Source(query).Do(context.TODO())
  201. if err != nil {
  202. log.Println("从ES查询出错", err.Error())
  203. return total, resNum, nil
  204. }
  205. total = searchResult.TotalHits()
  206. if searchResult.Hits != nil {
  207. resNum = len(searchResult.Hits.Hits)
  208. if isLimit && resNum < 5000 {
  209. res = make([]map[string]interface{}, resNum)
  210. for i, hit := range searchResult.Hits.Hits {
  211. //d := json.NewDecoder(bytes.NewBuffer(*hit.Source))
  212. //d.UseNumber()
  213. //d.Decode(&res[i])
  214. parseErr := json.Unmarshal(hit.Source, &res[i])
  215. if res[i] != nil {
  216. res[i]["_id"] = hit.Id
  217. }
  218. if isHighlight && parseErr == nil && hit.Highlight != nil && res[i] != nil {
  219. res[i]["highlight"] = map[string][]string(hit.Highlight)
  220. }
  221. }
  222. } else {
  223. log.Println("查询结果太多,查询到:", resNum, "条")
  224. }
  225. }
  226. }
  227. return total, resNum, &res
  228. }
  229. func GetOA(index, itype, query string) (*[]map[string]interface{}, int) {
  230. _, n, l := get(index, itype, query, "", true, true)
  231. return l, n
  232. }
  233. func GetNoLimit(index, itype, query string) *[]map[string]interface{} {
  234. _, _, l := get(index, itype, query, "", false, false)
  235. return l
  236. }
  237. //分页查询
  238. //{"name":"张三","$and":[{"age":{"$gt":10}},{"age":{"$lte":20}}]}
  239. //fields直接是 `"_id","title"`
  240. func GetPage(index, itype, query, order, field string, start, limit int) *[]map[string]interface{} {
  241. return Get(index, itype, MakeQuery(query, order, field, start, limit))
  242. }
  243. //openapi
  244. func GetOAPage(index, itype, query, order, field string, start, limit int) (*[]map[string]interface{}, int) {
  245. return GetOA(index, itype, MakeQuery(query, order, field, start, limit))
  246. }
  247. var SR = strings.Replace
  248. func MakeQuery(query, order, fileds string, start, limit int) string {
  249. res := AnalyQuery(query, "", QStr)
  250. if len(res) > 10 {
  251. res = SR(SR(SR(SR(res, ",$and", "", -1), "$and", "", -1), ",$or", "", -1), "$or", "", -1)
  252. if len(fileds) > 0 {
  253. //"_source":["account_number","balance"]
  254. res = res[:len(res)-1] + `,"_source":[` + fileds + "]}"
  255. }
  256. //{"name":-1,"age":1}
  257. if len(order) > 0 {
  258. res = res[:len(res)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  259. }
  260. if start > -1 {
  261. res = res[:len(res)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  262. }
  263. return res
  264. }
  265. return ""
  266. }
  267. //{"name":"aaa"}
  268. func AnalyQuery(query interface{}, parent string, result string) string {
  269. m := make(map[string]interface{})
  270. if q1, ok := query.(string); ok {
  271. json.Unmarshal([]byte(q1), &m)
  272. } else if q2, ok2 := query.(map[string]interface{}); ok2 {
  273. m = q2
  274. }
  275. if len(parent) == 0 {
  276. for k, v := range m {
  277. if k == "$and" || k == "$or" {
  278. temps := ""
  279. if map1, ok := v.([]interface{}); ok {
  280. for i := 0; i < len(map1); i++ {
  281. temps += "," + AnalyQuery(map1[i], k, "")
  282. }
  283. }
  284. if len(temps) > 0 {
  285. temps = temps[1:]
  286. }
  287. result = SR(result, k, temps+","+k, 1)
  288. } else {
  289. switch reflect.TypeOf(v).String() {
  290. case "string":
  291. if strings.Index(k, "TERM_") == 0 {
  292. result = SR(result, "$and", `{"term":{"`+SR(k, "TERM_", "", 1)+`":"`+fmt.Sprintf("%v", v)+`"}},$and`, 1)
  293. } else {
  294. result = SR(result, "$and", `{"query_string":{"default_field":"`+k+`","query":"`+fmt.Sprintf("%v", v)+`"}},$and`, 1)
  295. }
  296. case "int", "int8", "int32", "int64", "float32", "float64":
  297. if strings.Index(k, "TERM_") == 0 {
  298. result = SR(result, "$and", `{"term":{"`+SR(k, "TERM_", "", 1)+`":`+fmt.Sprintf("%v", v)+`}},$and`, 1)
  299. } else {
  300. result = SR(result, "$and", `{"query_string":{"default_field":"`+k+`","query":`+fmt.Sprintf("%v", v)+`}},$and`, 1)
  301. }
  302. default:
  303. result = SR(result, "$and", AnalyQuery(v, k, "")+",$and", 1)
  304. }
  305. }
  306. }
  307. return result
  308. } else {
  309. for k, v := range m {
  310. if k == "$in" {
  311. s := ""
  312. if map1, ok := v.([]interface{}); ok {
  313. for i := 0; i < len(map1); i++ {
  314. s += "," + `"` + fmt.Sprintf("%v", map1[i]) + `"`
  315. }
  316. }
  317. if len(s) > 0 {
  318. s = s[1:]
  319. }
  320. return `{"terms":{"` + parent + `":[` + s + `]}}`
  321. } else if strings.Contains(k, "$lt") || strings.Contains(k, "$gt") {
  322. return `{"range":{"` + parent + `":{"` + SR(k, "$", "", 1) + `":` + fmt.Sprintf("%v", v) + `}}}`
  323. } else {
  324. switch reflect.TypeOf(v).String() {
  325. case "string":
  326. if strings.Index(k, "TERM_") == 0 {
  327. return `{"term":{"` + SR(k, "TERM_", "", 1) + `":"` + fmt.Sprintf("%v", v) + `"}}`
  328. } else {
  329. return `{"query_string":{"default_field":"` + k + `","query":"` + fmt.Sprintf("%v", v) + `"}}`
  330. }
  331. case "int", "int8", "int32", "int64", "float32", "float64":
  332. if strings.Index(k, "TERM_") == 0 {
  333. return `{"term":{"` + SR(k, "TERM_", "", 1) + `":` + fmt.Sprintf("%v", v) + `}}`
  334. } else {
  335. return `{"query_string":{"default_field":"` + k + `","query":` + fmt.Sprintf("%v", v) + `}}`
  336. }
  337. default:
  338. return AnalyQuery(v, k, result)
  339. }
  340. }
  341. }
  342. }
  343. return result
  344. }
  345. func GetByIdField(index, itype, id, fields string) *map[string]interface{} {
  346. client := GetEsConn()
  347. defer DestoryEsConn(client)
  348. if client != nil {
  349. defer func() {
  350. if r := recover(); r != nil {
  351. log.Println("[E]", r)
  352. for skip := 1; ; skip++ {
  353. _, file, line, ok := runtime.Caller(skip)
  354. if !ok {
  355. break
  356. }
  357. go log.Printf("%v,%v\n", file, line)
  358. }
  359. }
  360. }()
  361. query := `{"query":{"term":{"_id":"` + id + `"}}`
  362. if len(fields) > 0 {
  363. query = query + `,"_source":[` + fields + `]`
  364. }
  365. query = query + "}"
  366. searchResult, err := client.Search().Index(index).Source(query).Do(context.TODO())
  367. if err != nil {
  368. log.Println("从ES查询出错", err.Error())
  369. return nil
  370. }
  371. var res map[string]interface{}
  372. if searchResult.Hits != nil {
  373. resNum := len(searchResult.Hits.Hits)
  374. if resNum == 1 {
  375. res = make(map[string]interface{})
  376. for _, hit := range searchResult.Hits.Hits {
  377. json.Unmarshal(hit.Source, &res)
  378. if res != nil {
  379. res["_id"] = hit.Id
  380. }
  381. }
  382. return &res
  383. }
  384. }
  385. }
  386. return nil
  387. }
  388. //根据id来查询文档
  389. func GetById(index, itype string, ids ...string) *[]map[string]interface{} {
  390. client := GetEsConn()
  391. defer DestoryEsConn(client)
  392. var res []map[string]interface{}
  393. if client != nil {
  394. defer func() {
  395. if r := recover(); r != nil {
  396. log.Println("[E]", r)
  397. for skip := 1; ; skip++ {
  398. _, file, line, ok := runtime.Caller(skip)
  399. if !ok {
  400. break
  401. }
  402. go log.Printf("%v,%v\n", file, line)
  403. }
  404. }
  405. }()
  406. query := es.NewIdsQuery().Ids(ids...)
  407. searchResult, err := client.Search().Index(index).Query(query).Do(context.TODO())
  408. if err != nil {
  409. log.Println("从ES查询出错", err.Error())
  410. return nil
  411. }
  412. if searchResult.Hits != nil {
  413. resNum := len(searchResult.Hits.Hits)
  414. if resNum < 5000 {
  415. res = make([]map[string]interface{}, resNum)
  416. for i, hit := range searchResult.Hits.Hits {
  417. json.Unmarshal(hit.Source, &res[i])
  418. if res[i] != nil {
  419. res[i]["_id"] = hit.Id
  420. }
  421. }
  422. } else {
  423. log.Println("查询结果太多,查询到:", resNum, "条")
  424. }
  425. }
  426. }
  427. return &res
  428. }
  429. //根据语句更新对象
  430. func Update(index, itype, id string, updateStr string) bool {
  431. client := GetEsConn()
  432. defer DestoryEsConn(client)
  433. b := false
  434. if client != nil {
  435. defer func() {
  436. if r := recover(); r != nil {
  437. log.Println("[E]", r)
  438. for skip := 1; ; skip++ {
  439. _, file, line, ok := runtime.Caller(skip)
  440. if !ok {
  441. break
  442. }
  443. go log.Printf("%v,%v\n", file, line)
  444. }
  445. }
  446. }()
  447. var err error
  448. esc := es.NewScript(updateStr)
  449. esc.Lang("groovy")
  450. _, err = client.Update().Index(index).Type(itype).Id(id).Script(esc).Do(context.TODO())
  451. if err != nil {
  452. log.Println("更新检索出错:", err.Error())
  453. } else {
  454. b = true
  455. }
  456. }
  457. return b
  458. }
  459. func BulkUpdate(index, itype string, ids []string, updateStr string) {
  460. client := GetEsConn()
  461. defer DestoryEsConn(client)
  462. if client != nil {
  463. defer func() {
  464. if r := recover(); r != nil {
  465. log.Println("[E]", r)
  466. for skip := 1; ; skip++ {
  467. _, file, line, ok := runtime.Caller(skip)
  468. if !ok {
  469. break
  470. }
  471. go log.Printf("%v,%v\n", file, line)
  472. }
  473. }
  474. }()
  475. for _, id := range ids {
  476. esc := es.NewScript(updateStr)
  477. esc.Lang("groovy")
  478. _, err := client.Update().Index(index).Type(itype).Id(id).Script(esc).Do(context.TODO())
  479. if err != nil {
  480. log.Println("更新检索出错:", err.Error())
  481. }
  482. }
  483. }
  484. }
  485. //根据id删除索引对象
  486. func DelById(index, itype, id string) bool {
  487. client := GetEsConn()
  488. defer DestoryEsConn(client)
  489. b := false
  490. if client != nil {
  491. defer func() {
  492. if r := recover(); r != nil {
  493. log.Println("[E]", r)
  494. for skip := 1; ; skip++ {
  495. _, file, line, ok := runtime.Caller(skip)
  496. if !ok {
  497. break
  498. }
  499. go log.Printf("%v,%v\n", file, line)
  500. }
  501. }
  502. }()
  503. var err error
  504. _, err = client.Delete().Index(index).Id(id).Do(context.TODO())
  505. if err != nil {
  506. log.Println("更新检索出错:", err.Error())
  507. } else {
  508. b = true
  509. }
  510. }
  511. return b
  512. }
  513. //先删除后增
  514. func UpdateNewDoc(index, itype string, obj ...interface{}) bool {
  515. client := GetEsConn()
  516. defer DestoryEsConn(client)
  517. b := false
  518. if client != nil {
  519. defer func() {
  520. if r := recover(); r != nil {
  521. log.Println("[E]", r)
  522. for skip := 1; ; skip++ {
  523. _, file, line, ok := runtime.Caller(skip)
  524. if !ok {
  525. break
  526. }
  527. go log.Printf("%v,%v\n", file, line)
  528. }
  529. }
  530. }()
  531. var err error
  532. for _, v := range obj {
  533. tempObj := objToMap(v)
  534. if tempObj == nil || len(*tempObj) == 0 {
  535. continue
  536. }
  537. id := bsonIdToSId((*tempObj)["_id"])
  538. (*tempObj)["id"] = id
  539. delete(*tempObj, "_id")
  540. if id != "" {
  541. client.Delete().Index(index).Id(id).Do(context.TODO())
  542. }
  543. _, err = client.Index().Index(index).Id(id).BodyJson(tempObj).Do(context.TODO())
  544. if err != nil {
  545. log.Println("保存到ES出错", err.Error())
  546. } else {
  547. b = true
  548. }
  549. }
  550. }
  551. return b
  552. }
  553. func BulkSave(index, itype string, obj *[]map[string]interface{}, isDelBefore bool) {
  554. client := GetEsConn()
  555. defer DestoryEsConn(client)
  556. if client != nil {
  557. defer func() {
  558. if r := recover(); r != nil {
  559. log.Println("[E]", r)
  560. for skip := 1; ; skip++ {
  561. _, file, line, ok := runtime.Caller(skip)
  562. if !ok {
  563. break
  564. }
  565. go log.Printf("%v,%v\n", file, line)
  566. }
  567. }
  568. }()
  569. req := client.Bulk()
  570. for _, v := range *obj {
  571. if v == nil || len(v) == 0 {
  572. continue
  573. }
  574. _id := bsonIdToSId(v["_id"])
  575. v["id"] = _id
  576. delete(v, "_id")
  577. if isDelBefore && _id != "" {
  578. req = req.Add(es.NewBulkDeleteRequest().Index(index).Id(_id))
  579. }
  580. req = req.Add(es.NewBulkIndexRequest().Index(index).Id(_id).Doc(v))
  581. }
  582. _, err := req.Do(context.TODO())
  583. if err != nil {
  584. log.Println("批量保存到ES出错", err.Error())
  585. }
  586. }
  587. }
  588. func Count(index, itype string, query interface{}) int64 {
  589. client := GetEsConn()
  590. defer DestoryEsConn(client)
  591. if client != nil {
  592. defer func() {
  593. if r := recover(); r != nil {
  594. log.Println("[E]", r)
  595. for skip := 1; ; skip++ {
  596. _, file, line, ok := runtime.Caller(skip)
  597. if !ok {
  598. break
  599. }
  600. go log.Printf("%v,%v\n", file, line)
  601. }
  602. }
  603. }()
  604. var n int64
  605. var err error
  606. if qs, ok := query.(string); ok {
  607. n, err = client.Count(index).BodyString(qs).Do(context.TODO())
  608. } else if qi, ok2 := query.(es.Query); ok2 {
  609. n, err = client.Count(index).Query(qi).Do(context.TODO())
  610. }
  611. if err != nil {
  612. log.Println("统计出错", err.Error())
  613. }
  614. return n
  615. }
  616. return 0
  617. }
  618. //ngram精确查询
  619. /*
  620. {
  621. "query": {
  622. "bool": {
  623. "should": [
  624. {
  625. "bool":{
  626. "must":[
  627. { "multi_match": {
  628. "query": "智能",
  629. "type": "phrase",
  630. "fields": [
  631. "title"
  632. ],
  633. "analyzer": "my_ngram"
  634. }
  635. },{
  636. "multi_match": {
  637. "query": "机器",
  638. "type": "phrase",
  639. "fields": [
  640. "title"
  641. ],
  642. "analyzer": "my_ngram"
  643. }
  644. },{
  645. "multi_match": {
  646. "query": "2016",
  647. "type": "phrase",
  648. "fields": [
  649. "title"
  650. ],
  651. "analyzer": "my_ngram"
  652. }
  653. }
  654. ]
  655. }
  656. },
  657. {
  658. "bool":{
  659. "must":[
  660. { "multi_match": {
  661. "query": "河南",
  662. "type": "phrase",
  663. "fields": [
  664. "title"
  665. ],
  666. "analyzer": "my_ngram"
  667. }
  668. },{
  669. "multi_match": {
  670. "query": "工商",
  671. "type": "phrase",
  672. "fields": [
  673. "title"
  674. ],
  675. "analyzer": "my_ngram"
  676. }
  677. },{
  678. "multi_match": {
  679. "query": "2016",
  680. "type": "phrase",
  681. "fields": [
  682. "title"
  683. ],
  684. "analyzer": "my_ngram"
  685. }
  686. }
  687. ]
  688. }
  689. }
  690. ],"minimum_should_match": 1
  691. }
  692. },
  693. "_source": [
  694. "_id",
  695. "title"
  696. ],
  697. "from": 0,
  698. "size": 10,
  699. "sort": [{
  700. "publishtime": "desc"
  701. }]
  702. }
  703. */
  704. //"2016+智能+办公,"河南+工商"
  705. //["2016+智能+办公","河南+工商"]
  706. //QStr = `{"query":{"bool":{should":[$or],"minimum_should_match" : 1}}}`
  707. //{"bool":{"must":[]}}
  708. //{"multi_match": {"query": "$word","type": "phrase", "fields": [$field],"analyzer": "my_ngram"}}
  709. //"highlight": {"pre_tags": [""],"post_tags": [""],"fields": {"detail": {"fragment_size": 1,"number_of_fragments": 1},"title": {"fragment_size": 1,"number_of_fragments": 1}}}
  710. const (
  711. //此处最后少一个},正好NgramStr取[1:]多一个}
  712. FilterQuery = `{"query": {"filtered": {"filter": {"bool": {"must": [%s]}},%s}}`
  713. NgramStr = `{"query":{"bool":{"must":[%s],"should":[%s],"minimum_should_match": 1}}}`
  714. NgramMust = `{"bool":{"must":[%s]}}`
  715. NgramMustAndNot = `{"bool":{"must":[%s],"must_not":[%s]}}`
  716. minq = `{"multi_match": {"query": "%s","type": "phrase", "fields": [%s]}}`
  717. HL = `"highlight": {"pre_tags": [""],"post_tags": [""],"fields": {%s}}`
  718. highlightStr = `%s: {"fragment_size": %d,"number_of_fragments": 1}`
  719. FilterQuery_New = `{"query":{"bool":{"must": [%s%s%s],"should":[]}}}`
  720. MatchQueryString = `{"match": {%s: { "query":"%s", "operator": "and"}}}`
  721. HL_New = `"highlight": {"pre_tags": ["<HL>"],"post_tags": ["<HL>"],"fields": {%s}}`
  722. //数据查询高亮标记2019-07-10
  723. HL_MP = `"highlight": {"pre_tags": ["<HL>"],"post_tags": ["</HL>"],"fields": {%s}}`
  724. ik_highlightStr = `%s: {"fragment_size": %d,"number_of_fragments": 1,"require_field_match": true}`
  725. IK_pre_tags = `<font class=\"es-highlight\">`
  726. IK_post_tags = `</font>`
  727. HL_IK = `"highlight": {"pre_tags": ["` + IK_pre_tags + `"],"post_tags": ["` + IK_post_tags + `"],"fields": {%s}}`
  728. )
  729. //替换了"号
  730. func GetNgramQuery(query interface{}, mustquery, findfields string) (qstr string) {
  731. var words []string
  732. if q, ok := query.(string); ok {
  733. if q != "" {
  734. words = strings.Split(q, ",")
  735. }
  736. } else if q, ok := query.([]string); ok {
  737. words = q
  738. } else if q, ok := query.([]interface{}); ok {
  739. words = objArrToStringArr(q)
  740. }
  741. if words != nil {
  742. new_minq := fmt.Sprintf(minq, "%s", findfields)
  743. musts := []string{}
  744. for _, qs_words := range words {
  745. qws := strings.Split(qs_words, "+")
  746. mq := []string{}
  747. for _, qs_word := range qws {
  748. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
  749. }
  750. musts = append(musts, fmt.Sprintf(NgramMust, strings.Join(mq, ",")))
  751. }
  752. qstr = fmt.Sprintf(NgramStr, mustquery, strings.Join(musts, ","))
  753. //log.Println("ngram-query", qstr)
  754. } else {
  755. qstr = fmt.Sprintf(NgramStr, mustquery, "")
  756. }
  757. return
  758. }
  759. func GetNgramQuery_New(querystring, querymust interface{}, must, findfields string) (qstring string) {
  760. querymust_string := ""
  761. var wordsMust []string
  762. if q, ok := querymust.(string); ok {
  763. if q != "" {
  764. wordsMust = strings.Split(q, ",")
  765. }
  766. } else if q, ok := querymust.([]string); ok {
  767. wordsMust = q
  768. } else if q, ok := querymust.([]interface{}); ok {
  769. wordsMust = objArrToStringArr(q)
  770. }
  771. if wordsMust != nil {
  772. new_minq := fmt.Sprintf(minq, "%s", findfields)
  773. musts := []string{}
  774. for _, qs_wordsMust := range wordsMust {
  775. qws := strings.Split(qs_wordsMust, "+")
  776. mq := []string{}
  777. for _, qs_word := range qws {
  778. mq = append(mq, fmt.Sprintf(new_minq, qs_word))
  779. }
  780. musts = append(musts, fmt.Sprintf(NgramMust, strings.Join(mq, ",")))
  781. }
  782. querymust_string = strings.Join(musts, ",")
  783. }
  784. //log.Println("must", must, querymust_string)
  785. //querystring---------------------------------------------
  786. query_string := ""
  787. var querysShold []string
  788. if q, ok := querystring.(string); ok {
  789. if q != "" {
  790. querysShold = strings.Split(q, ",")
  791. }
  792. } else if q, ok := querystring.([]string); ok {
  793. querysShold = q
  794. } else if q, ok := querystring.([]interface{}); ok {
  795. querysShold = objArrToStringArr(q)
  796. }
  797. if querysShold != nil {
  798. for k, name := range strings.Split(findfields, ",") {
  799. for _, qs_querysShold := range querysShold {
  800. if k > 0 {
  801. query_string = query_string + "," + fmt.Sprintf(MatchQueryString, fmt.Sprint(name), qs_querysShold)
  802. } else {
  803. query_string = query_string + fmt.Sprintf(MatchQueryString, fmt.Sprint(name), qs_querysShold)
  804. }
  805. }
  806. }
  807. }
  808. //log.Println("querystring", query_string)
  809. if querymust_string == "" {
  810. qstring = fmt.Sprintf(FilterQuery_New, must, query_string, querymust_string)
  811. } else {
  812. qstring = fmt.Sprintf(FilterQuery_New, must, query_string, ","+querymust_string)
  813. }
  814. return
  815. }
  816. func GetByNgram(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int) *[]map[string]interface{} {
  817. return GetByNgramAll(index, itype, query, mustquery, findfields, order, fields, start, limit, false, false)
  818. }
  819. //增加高亮、过滤查询、高亮截取字数
  820. func GetByNgramOther(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool, count int) *[]map[string]interface{} {
  821. defer catch()
  822. qstr := ""
  823. if mustquery != "" && filtermode {
  824. qstr = GetNgramQuery(query, "", findfields)
  825. qstr = fmt.Sprintf(FilterQuery, mustquery, qstr[1:])
  826. } else {
  827. qstr = GetNgramQuery(query, mustquery, findfields)
  828. }
  829. if qstr != "" {
  830. if highlight {
  831. ws := []string{}
  832. for _, w := range strings.Split(findfields, ",") {
  833. ws = append(ws, fmt.Sprintf(highlightStr, w, count))
  834. }
  835. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  836. }
  837. if len(fields) > 0 {
  838. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  839. }
  840. if len(order) > 0 {
  841. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  842. }
  843. if start > -1 {
  844. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  845. }
  846. //log.Println("ngram-find", qstr)
  847. return Get(index, itype, qstr)
  848. } else {
  849. return nil
  850. }
  851. }
  852. //增加高亮、过滤查询
  853. //替换了"号
  854. func GetByNgramAll(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool) *[]map[string]interface{} {
  855. defer catch()
  856. qstr := ""
  857. if mustquery != "" && filtermode {
  858. qstr = GetNgramQuery(query, "", findfields)
  859. qstr = fmt.Sprintf(FilterQuery, mustquery, qstr[1:])
  860. } else {
  861. qstr = GetNgramQuery(query, mustquery, findfields)
  862. }
  863. if qstr != "" {
  864. if highlight {
  865. ws := []string{}
  866. for _, w := range strings.Split(findfields, ",") {
  867. ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
  868. }
  869. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  870. }
  871. if len(fields) > 0 {
  872. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  873. }
  874. if strings.HasPrefix(order, "CUSTOM_") {
  875. qstr = qstr[:len(qstr)-1] + `,` + strings.TrimLeft(order, "CUSTOM_") + `}`
  876. } else if len(order) > 0 {
  877. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  878. }
  879. if start > -1 {
  880. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  881. }
  882. //log.Println("ngram-find", qstr)
  883. return Get(index, itype, qstr)
  884. } else {
  885. return nil
  886. }
  887. }
  888. //增加高亮、过滤查询
  889. func GetByNgramAll_New(index, itype string, querystring, querymust interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool) *[]map[string]interface{} {
  890. defer catch()
  891. qstr := ""
  892. if filtermode {
  893. qstr = GetNgramQuery_New(querystring, querymust, mustquery, findfields)
  894. } else {
  895. qstr = GetNgramQuery_New(querystring, "", mustquery, findfields)
  896. }
  897. if qstr != "" {
  898. if highlight {
  899. ws := []string{}
  900. for _, w := range strings.Split(findfields, ",") {
  901. ws = append(ws, w+`:{"force_source": true}`)
  902. }
  903. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_New, strings.Join(ws, ",")) + `}`
  904. }
  905. if len(fields) > 0 {
  906. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  907. }
  908. if len(order) > 0 {
  909. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", ",", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  910. }
  911. if start > -1 {
  912. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  913. }
  914. //log.Println("ngram-find", order, qstr)
  915. return Get(index, itype, qstr)
  916. } else {
  917. return nil
  918. }
  919. }
  920. type KeyConfig struct {
  921. Keys []string `json:"key"`
  922. NotKeys []string `json:"notkey"`
  923. InfoTypes []string `json:"infotype"`
  924. Areas []string `json:"area"`
  925. }
  926. //替换了"号
  927. func GetResForJY(index, itype string, keys []KeyConfig, allquery, findfields, SortQuery, fields string, start, limit int) *[]map[string]interface{} {
  928. if len(keys) > 0 {
  929. qstr := ""
  930. new_minq := fmt.Sprintf(minq, "%s", findfields)
  931. not_new_minq := fmt.Sprintf(minq, "%s", findfields) //排除词只查询标题
  932. musts := []string{}
  933. for _, qs_words := range keys {
  934. mq := []string{}
  935. notmq := []string{}
  936. for _, qs_word := range qs_words.Keys {
  937. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
  938. /*
  939. qs := AnalyzerWord("bidding", qs_word)
  940. for _, qw := range qs {
  941. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qw)))
  942. }
  943. */
  944. }
  945. for _, qs_word := range qs_words.NotKeys {
  946. notmq = append(notmq, fmt.Sprintf(not_new_minq, ReplaceYH(qs_word)))
  947. }
  948. if len(qs_words.Areas) > 0 {
  949. mq = append(mq, fmt.Sprintf(`{"terms":{"area":["%s"]}}`, strings.Join(qs_words.Areas, `","`)))
  950. }
  951. if len(qs_words.InfoTypes) > 0 {
  952. mq = append(mq, fmt.Sprintf(`{"terms":{"toptype":["%s"]}}`, strings.Join(qs_words.InfoTypes, `","`)))
  953. }
  954. musts = append(musts, fmt.Sprintf(NgramMustAndNot, strings.Join(mq, ","), strings.Join(notmq, ",")))
  955. }
  956. qstr = fmt.Sprintf(NgramStr, "", strings.Join(musts, ","))
  957. qstr = fmt.Sprintf(FilterQuery, allquery, qstr[1:])
  958. ws := []string{}
  959. for _, w := range strings.Split(findfields, ",") {
  960. ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
  961. }
  962. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  963. if len(fields) > 0 {
  964. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  965. }
  966. if len(SortQuery) > 0 {
  967. qstr = qstr[:len(qstr)-1] + `,"sort":` + SortQuery + `}`
  968. }
  969. if start > -1 {
  970. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  971. }
  972. //log.Println("jy-ngram-find", qstr)
  973. return Get(index, itype, qstr)
  974. } else {
  975. return nil
  976. }
  977. }
  978. func ReplaceYH(src string) (rpl string) {
  979. return strings.Replace(src, `"`, `\"`, -1)
  980. }
  981. //
  982. func GetAllByNgram(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
  983. if qstr != "" {
  984. if highlight {
  985. ws := []string{}
  986. for _, w := range strings.Split(findfields, ",") {
  987. ws = append(ws, fmt.Sprintf(highlightStr, w, count))
  988. }
  989. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  990. }
  991. if len(fields) > 0 {
  992. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  993. }
  994. if len(order) > 0 {
  995. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  996. }
  997. if start > -1 {
  998. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  999. }
  1000. log.Println("GetAllByNgram:", qstr)
  1001. return Get(index, itype, qstr)
  1002. } else {
  1003. return nil
  1004. }
  1005. }
  1006. //数据标记2019-07-10
  1007. func GetAllByNgram_MP(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
  1008. if qstr != "" {
  1009. if highlight {
  1010. ws := []string{}
  1011. for _, w := range strings.Split(findfields, ",") {
  1012. ws = append(ws, fmt.Sprintf(highlightStr, w, count))
  1013. }
  1014. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_MP, strings.Join(ws, ",")) + `}`
  1015. }
  1016. if len(fields) > 0 {
  1017. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  1018. }
  1019. if len(order) > 0 {
  1020. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  1021. }
  1022. if start > -1 {
  1023. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1024. }
  1025. // log.Println("GetAllByNgram:", qstr)
  1026. return Get(index, itype, qstr)
  1027. } else {
  1028. return nil
  1029. }
  1030. }
  1031. //ik 分词
  1032. func GetAllByIk(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
  1033. if qstr != "" {
  1034. if highlight {
  1035. ws := []string{}
  1036. for _, w := range strings.Split(findfields, ",") {
  1037. ws = append(ws, fmt.Sprintf(ik_highlightStr, w, count))
  1038. }
  1039. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_IK, strings.Join(ws, ",")) + `}`
  1040. }
  1041. if len(fields) > 0 {
  1042. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  1043. }
  1044. if len(order) > 0 {
  1045. qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
  1046. }
  1047. if start > -1 {
  1048. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1049. }
  1050. //log.Println("GetAllByNgram:", qstr)
  1051. return Get(index, itype, qstr)
  1052. } else {
  1053. return nil
  1054. }
  1055. }
  1056. func GetResForJYView(index, itype string, keys []KeyConfig, allquery, findfields, SortQuery, fields string, start, limit int) *[]map[string]interface{} {
  1057. if len(keys) > 0 {
  1058. qstr := ""
  1059. new_minq := fmt.Sprintf(minq, "%s", findfields)
  1060. not_new_minq := fmt.Sprintf(minq, "%s", findfields) //排除词只查询标题
  1061. musts := []string{}
  1062. for _, qs_words := range keys {
  1063. mq := []string{}
  1064. notmq := []string{}
  1065. for _, qs_word := range qs_words.Keys {
  1066. mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
  1067. }
  1068. for _, qs_word := range qs_words.NotKeys {
  1069. notmq = append(notmq, fmt.Sprintf(not_new_minq, ReplaceYH(qs_word)))
  1070. }
  1071. if len(qs_words.Areas) > 0 {
  1072. mq = append(mq, fmt.Sprintf(`{"terms":{"area":["%s"]}}`, strings.Join(qs_words.Areas, `","`)))
  1073. }
  1074. if len(qs_words.InfoTypes) > 0 {
  1075. mq = append(mq, fmt.Sprintf(`{"terms":{"toptype":["%s"]}}`, strings.Join(qs_words.InfoTypes, `","`)))
  1076. }
  1077. musts = append(musts, fmt.Sprintf(NgramMustAndNot, strings.Join(mq, ","), strings.Join(notmq, ",")))
  1078. }
  1079. qstr = fmt.Sprintf(NgramStr, "", strings.Join(musts, ","))
  1080. qstr = fmt.Sprintf(FilterQuery, allquery, qstr[1:])
  1081. ws := []string{}
  1082. for _, w := range strings.Split(findfields, ",") {
  1083. ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
  1084. }
  1085. qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
  1086. if len(fields) > 0 {
  1087. qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
  1088. }
  1089. if len(SortQuery) > 0 {
  1090. qstr = qstr[:len(qstr)-1] + `,"sort":` + SortQuery + `}`
  1091. }
  1092. if start > -1 {
  1093. qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
  1094. }
  1095. return Get(index, itype, qstr)
  1096. } else {
  1097. return nil
  1098. }
  1099. }