elasticutil.go 32 KB

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