oprd_data/es/elasticutil.go

package main

import (
	"context"
	"runtime"
	"sync"
	"time"
	//	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"qfw/util"
	"qfw/util/mongodb"
	mongodbutil "qfw/util/mongodbutil"
	"reflect"

	es "gopkg.in/olivere/elastic.v5"

	"strconv"
	"strings"
)

//检索库服务地址
var (
	addrs     []string
	LocCity   = map[string]string{}
	SIZE      = 15
	pool      chan *es.Client
	ntimeout  int
	poolsize  = int32(20)
	newesLock = &sync.Mutex{}
	SR        = strings.Replace
)

const (
	QStr = `{"query":{"bool":{"must":[$and],"must_not":[],
	"should":[$or],"minimum_should_match" : 0}}}`
)

func InitElastic(addr string) {
	InitElasticSize(addr, 5)
}

//n倍的池
func InitElasticSize(addr string, size int) {
	poolsize = int32(3 * size)
	pool = make(chan *es.Client, poolsize)
	for _, s := range strings.Split(addr, ",") {
		addrs = append(addrs, s)
	}
	for i := 0; i < size; i++ {
		client, _ := es.NewClient(es.SetURL(addrs...), es.SetMaxRetries(2), es.SetSniff(false))
		pool <- client
	}
}

//关闭连接
func DestoryEsConn(client *es.Client) {
	select {
	case pool <- client:
		break
	case <-time.After(time.Second * 1):
		if client != nil {
			client.Stop()
		}
		client = nil
	}
}

var (
	lastTime     = int64(0)
	lastTimeLock = &sync.Mutex{}
)

//获取连接
func GetEsConn() *es.Client {
	select {
	case c := <-pool:
		if c == nil || !c.IsRunning() {
			log.Println("new esclient.", len(pool))
			client, err := es.NewClient(es.SetURL(addrs...),
				es.SetMaxRetries(2), es.SetSniff(false))
			if err == nil && client.IsRunning() {
				return client
			}
		}
		return c
	case <-time.After(time.Second * 4):
		//超时
		ntimeout++
		lastTimeLock.Lock()
		defer lastTimeLock.Unlock()
		//12秒后允许创建链接
		c := time.Now().Unix() - lastTime
		if c > 6 {
			lastTime = time.Now().Unix()
			log.Println("add client..", len(pool))
			c, _ := es.NewClient(es.SetURL(addrs...), es.SetMaxRetries(2), es.SetSniff(false))
			go func() {
				for i := 0; i < 2; i++ {
					client, _ := es.NewClient(es.SetURL(addrs...), es.SetMaxRetries(2), es.SetSniff(false))
					pool <- client
				}
			}()
			return c
		}
		return nil
	}
}

//保存对象
func Save(index, itype string, obj interface{}) bool {
	client := GetEsConn()
	defer DestoryEsConn(client)
	defer func() {
		if r := recover(); r != nil {
			log.Println("[E]", r)
			for skip := 1; ; skip++ {
				_, file, line, ok := runtime.Caller(skip)
				if !ok {
					break
				}
				go log.Printf("%v,%v\n", file, line)
			}
		}
	}()
	data := util.ObjToMap(obj)
	_id := util.BsonIdToSId((*data)["_id"])
	(*data)["id"] = _id
	delete((*data), "_id")
	_, err := client.Index().Index(index).Type(itype).Id(_id).BodyJson(data).Do(context.TODO())
	if err != nil {
		log.Println("保存到ES出错", err.Error(), obj)
		return false
	} else {
		return true
	}

}

//底层查询方法
func Get(index, itype, query string) *[]map[string]interface{} {
	//log.Println("query  -- ", query)
	client := GetEsConn()
	defer DestoryEsConn(client)
	var res []map[string]interface{}
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		searchResult, err := client.Search().Index(index).Type(itype).Source(query).Do(context.TODO())
		if err != nil {
			log.Println("从ES查询出错", err.Error())
			return nil
		}
		if searchResult.Hits != nil {
			resNum := len(searchResult.Hits.Hits)
			if resNum < 3000 {
				res = make([]map[string]interface{}, resNum)
				for i, hit := range searchResult.Hits.Hits {
					parseErr := json.Unmarshal(*hit.Source, &res[i])
					if res[i] != nil {
						res[i]["_id"] = hit.Id
					}
					if parseErr == nil && hit.Highlight != nil && res[i] != nil {
						res[i]["highlight"] = map[string][]string(hit.Highlight)
					}
				}
			} else {
				log.Println("查询结果太多，查询到：", resNum, "条")
			}
		}
	}
	return &res
}

func GetNoLimit(index, itype, query string) *[]map[string]interface{} {
	client := GetEsConn()
	defer DestoryEsConn(client)
	var res []map[string]interface{}
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		searchResult, err := client.Search().Index(index).Type(itype).Source(query).Do(context.TODO())
		if err != nil {
			log.Println("从ES查询出错", err.Error())
			return nil
		}

		if searchResult.Hits != nil {
			resNum := len(searchResult.Hits.Hits)
			res = make([]map[string]interface{}, resNum)
			for i, hit := range searchResult.Hits.Hits {
				json.Unmarshal(*hit.Source, &res[i])
				if res[i] != nil {
					res[i]["_id"] = hit.Id
				}
			}
		}
	}
	return &res
}

//分页查询
//{"name":"张三","$and":[{"age":{"$gt":10}},{"age":{"$lte":20}}]}
//fields直接是 `"_id","title"`
func GetPage(index, itype, query, order, field string, start, limit int) *[]map[string]interface{} {
	return Get(index, itype, MakeQuery(query, order, field, start, limit))
}

func MakeQuery(query, order, fileds string, start, limit int) string {
	res := AnalyQuery(query, "", QStr)
	if len(res) > 10 {
		res = SR(SR(SR(SR(res, ",$and", "", -1), "$and", "", -1), ",$or", "", -1), "$or", "", -1)
		if len(fileds) > 0 {
			//"_source":["account_number","balance"]
			res = res[:len(res)-1] + `,"_source":[` + fileds + "]}"
		}
		//{"name":-1,"age":1}
		if len(order) > 0 {
			res = res[:len(res)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
		}
		if start > -1 {
			res = res[:len(res)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
		}
		return res
	}
	return ""
}

//{"name":"aaa"}
func AnalyQuery(query interface{}, parent string, result string) string {
	m := make(map[string]interface{})
	if q1, ok := query.(string); ok {
		json.Unmarshal([]byte(q1), &m)
	} else if q2, ok2 := query.(map[string]interface{}); ok2 {
		m = q2
	}
	if len(parent) == 0 {
		for k, v := range m {
			if k == "$and" || k == "$or" {
				temps := ""
				if map1, ok := v.([]interface{}); ok {
					for i := 0; i < len(map1); i++ {
						temps += "," + AnalyQuery(map1[i], k, "")
					}
				}
				if len(temps) > 0 {
					temps = temps[1:]
				}
				result = SR(result, k, temps+","+k, 1)
			} else {
				switch reflect.TypeOf(v).String() {
				case "string":
					if strings.Index(k, "TERM_") == 0 {
						result = SR(result, "$and", `{"term":{"`+SR(k, "TERM_", "", 1)+`":"`+fmt.Sprintf("%v", v)+`"}},$and`, 1)
					} else {
						result = SR(result, "$and", `{"query_string":{"default_field":"`+k+`","query":"`+fmt.Sprintf("%v", v)+`"}},$and`, 1)
					}
				case "int", "int8", "int32", "int64", "float32", "float64":
					if strings.Index(k, "TERM_") == 0 {
						result = SR(result, "$and", `{"term":{"`+SR(k, "TERM_", "", 1)+`":`+fmt.Sprintf("%v", v)+`}},$and`, 1)
					} else {
						result = SR(result, "$and", `{"query_string":{"default_field":"`+k+`","query":`+fmt.Sprintf("%v", v)+`}},$and`, 1)
					}
				default:
					result = SR(result, "$and", AnalyQuery(v, k, "")+",$and", 1)
				}
			}
		}
		return result
	} else {
		for k, v := range m {
			if k == "$in" {
				s := ""
				if map1, ok := v.([]interface{}); ok {
					for i := 0; i < len(map1); i++ {
						s += "," + `"` + fmt.Sprintf("%v", map1[i]) + `"`
					}
				}
				if len(s) > 0 {
					s = s[1:]
				}
				return `{"terms":{"` + parent + `":[` + s + `]}}`
			} else if strings.Contains(k, "$lt") || strings.Contains(k, "$gt") {
				return `{"range":{"` + parent + `":{"` + SR(k, "$", "", 1) + `":` + fmt.Sprintf("%v", v) + `}}}`
			} else {
				switch reflect.TypeOf(v).String() {
				case "string":
					if strings.Index(k, "TERM_") == 0 {
						return `{"term":{"` + SR(k, "TERM_", "", 1) + `":"` + fmt.Sprintf("%v", v) + `"}}`
					} else {
						return `{"query_string":{"default_field":"` + k + `","query":"` + fmt.Sprintf("%v", v) + `"}}`
					}
				case "int", "int8", "int32", "int64", "float32", "float64":
					if strings.Index(k, "TERM_") == 0 {
						return `{"term":{"` + SR(k, "TERM_", "", 1) + `":` + fmt.Sprintf("%v", v) + `}}`
					} else {
						return `{"query_string":{"default_field":"` + k + `","query":` + fmt.Sprintf("%v", v) + `}}`
					}
				default:
					return AnalyQuery(v, k, result)
				}
			}
		}
	}
	return result
}

func GetByIdField(index, itype, id, fields string) *map[string]interface{} {
	client := GetEsConn()
	defer DestoryEsConn(client)
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		query := `{"query":{"term":{"_id":"` + id + `"}}`
		if len(fields) > 0 {
			query = query + `,"_source":[` + fields + `]`
		}
		query = query + "}"
		searchResult, err := client.Search().Index(index).Type(itype).Source(query).Do(context.TODO())
		if err != nil {
			log.Println("从ES查询出错", err.Error())
			return nil
		}
		var res map[string]interface{}
		if searchResult.Hits != nil {
			resNum := len(searchResult.Hits.Hits)
			if resNum == 1 {
				res = make(map[string]interface{})
				for _, hit := range searchResult.Hits.Hits {
					json.Unmarshal(*hit.Source, &res)
					if res != nil {
						res["_id"] = hit.Id
					}
				}
				return &res
			} else {
				log.Println("查询结果太多，查询到：", resNum, "条")
			}

		}
	}
	return nil
}

//根据id来查询文档
func GetById(index, itype string, ids ...string) *[]map[string]interface{} {
	client := GetEsConn()
	defer DestoryEsConn(client)
	var res []map[string]interface{}
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		query := es.NewIdsQuery().Ids(ids...)
		searchResult, err := client.Search().Index(index).Type(itype).Query(query).Do(context.TODO())
		if err != nil {
			log.Println("从ES查询出错", err.Error())
			return nil
		}

		if searchResult.Hits != nil {
			resNum := len(searchResult.Hits.Hits)
			if resNum < 5000 {
				res = make([]map[string]interface{}, resNum)
				for i, hit := range searchResult.Hits.Hits {
					json.Unmarshal(*hit.Source, &res[i])
					if res[i] != nil {
						res[i]["_id"] = hit.Id
					}
				}
			} else {
				log.Println("查询结果太多，查询到：", resNum, "条")
			}

		}
	}
	return &res
}

//elastic_v5之后查询语句适配
func StringToQuery(q string) es.Query {
	defer util.Catch()
	m := map[string]interface{}{}
	err := json.Unmarshal([]byte(q), &m)
	if err == nil {
		q, _ := m["query"].(map[string]interface{})
		return AQ(q)
	}
	return nil
}

func AQ(m map[string]interface{}) (bq es.Query) {
	for k, v := range m {
		if k == "bool" {
			bq1 := es.NewBoolQuery()
			b1, _ := v.(map[string]interface{})
			if b1["must"] != nil {
				bm, _ := b1["must"].([]interface{})
				for _, tm := range bm {
					tm1, _ := tm.(map[string]interface{})
					if tm1 != nil {
						bq1.Must(AQ(tm1))
					}
				}
			}
			if b1["must_not"] != nil {
				bm, _ := b1["must_not"].([]interface{})
				for _, tm := range bm {
					tm1, _ := tm.(map[string]interface{})
					if tm1 != nil {
						bq1.MustNot(AQ(tm1))
					}
				}
			}
			if b1["should"] != nil {
				bm, _ := b1["should"].([]interface{})
				for _, tm := range bm {
					tm1, _ := tm.(map[string]interface{})
					if tm1 != nil {
						bq1.Should(AQ(tm1))
					}
				}
			}
			if b1["minimum_should_match"] != nil {
				bq1.MinimumNumberShouldMatch(util.IntAll(b1["minimum_should_match"]))
			}
			bq = bq1
		} else if k == "term" {
			b1, _ := v.(map[string]interface{})
			for k, v := range b1 {
				bq = es.NewTermQuery(k, v)
				break
			}
		} else if k == "terms" {
			b1, _ := v.(map[string]interface{})
			for k, v := range b1 {
				vs, _ := v.([]interface{})
				bq = es.NewTermsQuery(k, vs...)
				break
			}
		} else if k == "range" {
			b1, _ := v.(map[string]interface{})
			for k, v := range b1 {
				vm := v.(map[string]interface{})
				bq1 := es.NewRangeQuery(k)
				if vm["$gt"] != nil {
					bq1.Gt(vm["$gt"])
				}
				if vm["$gte"] != nil {
					bq1.Gte(vm["$gte"])
				}
				if vm["$lt"] != nil {
					bq1.Lt(vm["$lt"])
				}
				if vm["$lte"] != nil {
					bq1.Lte(vm["$lte"])
				}
				break
			}
		}
	}
	return
}

//删除某个索引，根据查询
func Del(index, itype string, query interface{}) bool {
	client := GetEsConn()
	defer DestoryEsConn(client)
	b := false
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		var err error
		if qs, ok := query.(string); ok {
			//by, e := json.Marshal(qs)
			esq := StringToQuery(qs)
			if esq != nil {
				_, err = client.DeleteByQuery().Index(index).Type(itype).Query(esq).Do(context.TODO())
			}
		} else if qi, ok2 := query.(es.Query); ok2 {
			_, err = client.DeleteByQuery().Index(index).Type(itype).Query(qi).Do(context.TODO())
		}
		if err != nil {
			log.Println("删除索引出错:", err.Error())
		} else {
			b = true
		}
	}
	return b
}

//根据语句更新对象
func Update(index, itype, id string, updateStr string) bool {
	client := GetEsConn()
	defer DestoryEsConn(client)
	b := false
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		var err error
		esc := es.NewScript(updateStr)
		esc.Lang("groovy")
		_, err = client.Update().Index(index).Type(itype).Id(id).Script(esc).Do(context.TODO())
		if err != nil {
			log.Println("更新检索出错:", err.Error())
		} else {
			b = true
		}
	}
	return b
}

func BulkUpdate(index, itype string, ids []string, updateStr string) {
	client := GetEsConn()
	defer DestoryEsConn(client)
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		for _, id := range ids {
			esc := es.NewScript(updateStr)
			esc.Lang("groovy")
			_, err := client.Update().Index(index).Type(itype).Id(id).Script(esc).Do(context.TODO())
			if err != nil {
				log.Println("更新检索出错:", err.Error())
			}
		}
	}
}

//根据id删除索引对象
func DelById(index, itype, id string) bool {
	client := GetEsConn()
	defer DestoryEsConn(client)
	b := false
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		var err error
		_, err = client.Delete().Index(index).Type(itype).Id(id).Do(context.TODO())
		if err != nil {
			log.Println("更新检索出错:", err.Error())
		} else {
			b = true
		}
	}
	return b
}

//先删除后增
func UpdateNewDoc(index, itype string, obj ...interface{}) bool {
	client := GetEsConn()
	defer DestoryEsConn(client)
	b := false
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		var err error
		for _, v := range obj {
			tempObj := util.ObjToMap(v)
			if tempObj == nil || len(*tempObj) == 0 {
				continue
			}
			id := util.BsonIdToSId((*tempObj)["_id"])
			(*tempObj)["id"] = id
			delete(*tempObj, "_id")
			if id != "" {
				client.Delete().Index(index).Type(itype).Id(id).Do(context.TODO())
			}
			_, err = client.Index().Index(index).Type(itype).Id(id).BodyJson(tempObj).Do(context.TODO())
			if err != nil {
				log.Println("保存到ES出错", err.Error())
			} else {
				b = true
			}
		}

	}
	return b
}

func UpdateEntDoc(id string) bool {
	b := false
	map2 := map[string]interface{}{}
	util.ReadConfig(&map2)
	ent := mongodbutil.FindById("enterprise", map2["entMongodbAlias"].(string), map2["entMongodbName"].(string), id, "")
	_ent := mongodb.FindById("enterprise", id, "")
	if _ent != nil && len(*_ent) > 0 {
		for k, v := range *_ent {
			(*ent)[k] = v
		}
	}
	if ent != nil {
		b = UpdateNewDoc("enterprise", "enterprise", ConverData(ent))
	}
	return b
}

//把地市代码转为地市
func getLoc(code string, res *map[string]string) (loc string) {
	switch len(code) {
	case 6:
		loc = (*res)[code[:2]] + " " + (*res)[code[:4]] + " " + (*res)[code]
		break
	case 4:
		loc = (*res)[code[:2]] + " " + (*res)[code]
		break
	case 2:
		loc = (*res)[code]
		break
	}
	return
}

//把地市代码转为地市
func Loop(m interface{}, res *map[string]string) {
	m1, ok := m.([]interface{})
	if !ok {
		m2, _ := m.([]map[string]interface{})
		for i := 0; i < len(m2); i++ {
			ms := m2[i]
			(*res)[fmt.Sprintf("%1.0f", ms["k"])] = fmt.Sprintf("%s", ms["n"])
			s := ms["s"]
			if nil != s {
				mss, _ := s.([]interface{})
				if nil != mss {
					Loop(mss, res)
				}
			}
		}
	} else {
		for i := 0; i < len(m1); i++ {
			ms, _ := m1[i].(map[string]interface{})
			(*res)[fmt.Sprintf("%1.0f", ms["k"])] = fmt.Sprintf("%s", ms["n"])
			s := ms["s"]
			if nil != s {
				mss, _ := s.([]interface{})
				if nil != mss {
					Loop(mss, res)
				}
			}
		}
	}
}

func ConverData(ent *map[string]interface{}) map[string]interface{} {
	tmp := *ent
	id64, _ := tmp["ID"].(int64)
	ids := fmt.Sprintf("%d", id64)
	tmp2 := make(map[string]interface{})
	tmp2["ID"] = ids
	tmp2["_id"] = tmp["_id"]
	tmp2["id"] = tmp["_id"]
	tmp2["Area"] = tmp["Area"]
	tmp2["LeRep"] = tmp["LeRep"]
	tmp2["RegNo"] = tmp["RegNo"]
	tmp2["EntType"] = tmp["EntType"]
	tmp2["EntName"] = tmp["EntName"]
	tmp2["EntTypeName"] = tmp["EntTypeName"]
	tmp2["Dom"] = tmp["Dom"]
	tmp2["EstDate"] = tmp["EstDate"]
	tmp2["OpStateName"] = tmp["OpStateName"]
	tmp2["OpScope"] = tmp["OpScope"]
	tmp2["OpState"] = tmp["OpState"]
	tmp2["s_submitid"] = tmp["s_submitid"]
	tmp2["l_submittime"] = tmp["l_submittime"]
	tmp2["s_submitname"] = tmp["s_submitname"]
	tmp2["RegCapCurName"] = tmp["RegCapCurName"]
	//增加营业状态排序
	if tmp2["OpState"] == "06" {
		tmp2["OpSint"] = true
	} else {
		tmp2["OpSint"] = false
	}
	tmp2["OpLocDistrict"] = tmp["OpLocDistrict"]
	//增加代码转名称
	tmpLoc, _ := tmp["OpLocDistrict"].(string)
	tmp2["OpLocDistrictName"] = getLoc(tmpLoc, &LocCity)

	tmp2["RecCap"] = tmp["RecCap"]
	tmp2["RegCap"] = tmp["RegCap"]
	tmp2["IndustryPhy"] = tmp["IndustryPhy"]
	tmp2["IndustryPhyName"] = tmp["IndustryPhyName"]
	tmp2["RegOrg"] = tmp["RegOrg"]
	tmp2["RegOrgName"] = tmp["RegOrgName"]
	tmp2["Tel"] = tmp["Tel"]
	tmp2["CompForm"] = tmp["CompForm"]
	tmp2["CompFormName"] = tmp["CompFormName"]
	//增加异常名录标记 Ycml可能是bool也可能是string
	Ycmlb, _ := tmp["Ycml"].(bool)
	Ycmls, _ := tmp["Ycml"].(string)
	if Ycmlb || Ycmls == "1" {
		tmp2["Ycml"] = true
	} else {
		tmp2["Ycml"] = false
	}
	//增加年报联系信息
	if tmp["Nb_email"] != nil {
		tmp2["Nb_email"] = tmp["Nb_email"]
	}
	if tmp["Nb_tel"] != nil {
		tmp2["Nb_tel"] = tmp["Nb_tel"]
	}
	if tmp["Nb_addr"] != nil {
		tmp2["Nb_addr"] = tmp["Nb_addr"]
	}

	s_synopsis := tmp["s_synopsis"]
	if s_synopsis == nil {
		s_synopsis = ""
	}
	tmp2["s_synopsis"] = s_synopsis //企业简介

	//股东
	stock := getStock(tmp["investor"])
	tmp2["stock"] = stock

	tmp2["LegCerNO"] = tmp["LegCerNO"]
	if tmp["s_microwebsite"] != nil {
		tmp2["s_microwebsite"] = tmp["s_microwebsite"]
	}

	tmp2["SourceType"] = tmp["SourceType"] //数据来源
	s_servicenames := tmp["s_servicenames"]
	if s_servicenames == nil {
		s_servicenames = ""
	}
	tmp2["s_servicenames"] = s_servicenames //服务名称
	s_action := tmp["s_action"]
	if s_action == nil {
		s_action = "N"
	}
	tmp2["s_action"] = s_action
	tmp2["s_persion"] = tmp["s_persion"]
	tmp2["s_mobile"] = tmp["s_mobile"]
	tmp2["s_enturl"] = tmp["s_enturl"]
	tmp2["s_weixin"] = tmp["s_weixin"]
	tmp2["s_avatar"] = tmp["s_avatar"]
	return tmp2
}

func getStock(obj interface{}) string {
	stock := ""
	if ns, ok := obj.([]interface{}); ok {
		stock = " "
		for _, ns1 := range ns {
			if nn, ok1 := ns1.(map[string]interface{}); ok1 {
				tmp := fmt.Sprintf("%s", nn["Inv"])
				if strings.Index(stock, tmp) < 0 {
					stock += tmp + " "
				}
			}
		}
	}
	return stock
}

func BulkSave(index, itype string, obj *[]map[string]interface{}, isDelBefore bool) {
	client := GetEsConn()
	defer DestoryEsConn(client)
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		req := client.Bulk()
		for _, v := range *obj {
			if v == nil || len(v) == 0 {
				continue
			}
			_id := util.BsonIdToSId(v["_id"])
			v["id"] = _id
			delete(v, "_id")
			if isDelBefore && _id != "" {
				req = req.Add(es.NewBulkDeleteRequest().Index(index).Type(itype).Id(_id))
			}
			req = req.Add(es.NewBulkIndexRequest().Index(index).Type(itype).Id(_id).Doc(v))
		}
		_, err := req.Do(context.TODO())
		if err != nil {
			log.Println("批量保存到ES出错", err.Error())
		}
	}
}

func Count(index, itype string, query interface{}) int64 {
	client := GetEsConn()
	defer DestoryEsConn(client)
	if client != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Println("[E]", r)
				for skip := 1; ; skip++ {
					_, file, line, ok := runtime.Caller(skip)
					if !ok {
						break
					}
					go log.Printf("%v,%v\n", file, line)
				}
			}
		}()
		var qq es.Query
		if qs, ok := query.(string); ok {
			temp := &es.BoolQuery{
				QueryStrings: qs,
			}
			qq = temp
		} else if qi, ok2 := query.(es.Query); ok2 {
			qq = qi
		}
		n, err := client.Count(index).Type(itype).Query(qq).Do(context.TODO())
		if err != nil {
			log.Println("统计出错", err.Error())
		}

		return n
	}
	return 0
}

//ngram精确查询
/*
{
  "query": {
    "bool": {
      "should": [
        {
	"bool":{
	  "must":[
	  {     "multi_match": {
            "query": "智能",
            "type": "phrase",
            "fields": [
              "title"
            ],
           "analyzer": "my_ngram"
          }
        },{
          "multi_match": {
            "query": "机器",
            "type": "phrase",
            "fields": [
              "title"
            ],
           "analyzer": "my_ngram"
          }
        },{
          "multi_match": {
            "query": "2016",
            "type": "phrase",
            "fields": [
              "title"
            ],
           "analyzer": "my_ngram"
          }
	  }
	  ]
	}
        },

{
	"bool":{
	  "must":[
	  {          "multi_match": {
            "query": "河南",
            "type": "phrase",
            "fields": [
              "title"
            ],
           "analyzer": "my_ngram"
          }
        },{
          "multi_match": {
            "query": "工商",
            "type": "phrase",
            "fields": [
              "title"
            ],
           "analyzer": "my_ngram"
          }
        },{
          "multi_match": {
            "query": "2016",
            "type": "phrase",
            "fields": [
              "title"
            ],
           "analyzer": "my_ngram"
          }
	  }
	  ]
	}
        }
      ],"minimum_should_match": 1
    }
  },
  "_source": [
    "_id",
    "title"
  ],
  "from": 0,
  "size": 10,
  "sort": [{
      "publishtime": "desc"
    }]
}

*/
//"2016+智能+办公,"河南+工商"
//["2016+智能+办公","河南+工商"]
//QStr = `{"query":{"bool":{should":[$or],"minimum_should_match" : 1}}}`
//{"bool":{"must":[]}}
//{"multi_match": {"query": "$word","type": "phrase", "fields": [$field],"analyzer": "my_ngram"}}
//"highlight": {"pre_tags": [""],"post_tags": [""],"fields": {"detail": {"fragment_size": 1,"number_of_fragments": 1},"title": {"fragment_size": 1,"number_of_fragments": 1}}}
const (
	//此处最后少一个}，正好NgramStr取[1:]多一个}
	FilterQuery     = `{"query": {"bool": {"filter": [%s]}},%s`
	NgramStr        = `{"query":{"bool":{"must":[%s],"should":[%s],"minimum_should_match" : 0}}}`
	NgramMust       = `{"bool":{"must":[%s]}}`
	NgramMustAndNot = `{"bool":{"must":[%s],"must_not":[%s]}}`
	minq            = `{"multi_match": {"query": "%s","type": "phrase", "fields": [%s],"analyzer": "my_ngram"}}`
	HL              = `"highlight": {"pre_tags": [""],"post_tags": [""],"fields": {%s}}`
	highlightStr    = `%s: {"fragment_size": %d,"number_of_fragments": 1}`

	FilterQuery_New  = `{"query":{"bool":{"must": [%s%s%s],"should":[]}}}`
	MatchQueryString = `{"match": {%s: { "query":"%s", "operator": "and"}}}`
	HL_New           = `"highlight": {"pre_tags": ["<HL>"],"post_tags": ["<HL>"],"fields": {%s}}`
)

//替换了"号
func GetNgramQuery(query interface{}, mustquery, findfields string) (qstr string) {
	var words []string
	if q, ok := query.(string); ok {
		if q != "" {
			words = strings.Split(q, ",")
		}
	} else if q, ok := query.([]string); ok {
		words = q
	} else if q, ok := query.([]interface{}); ok {
		words = util.ObjArrToStringArr(q)
	}
	if words != nil {
		new_minq := fmt.Sprintf(minq, "%s", findfields)
		musts := []string{}
		for _, qs_words := range words {
			qws := strings.Split(qs_words, "+")
			mq := []string{}
			for _, qs_word := range qws {
				mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
			}
			musts = append(musts, fmt.Sprintf(NgramMust, strings.Join(mq, ",")))
		}
		qstr = fmt.Sprintf(NgramStr, mustquery, strings.Join(musts, ","))
		//log.Println("ngram-query", qstr)
	} else {
		qstr = fmt.Sprintf(NgramStr, mustquery, "")
	}
	return
}

func GetNgramQuery_New(querystring, querymust interface{}, must, findfields string) (qstring string) {
	querymust_string := ""
	var wordsMust []string
	if q, ok := querymust.(string); ok {
		if q != "" {
			wordsMust = strings.Split(q, ",")
		}
	} else if q, ok := querymust.([]string); ok {
		wordsMust = q
	} else if q, ok := querymust.([]interface{}); ok {
		wordsMust = util.ObjArrToStringArr(q)
	}
	if wordsMust != nil {
		new_minq := fmt.Sprintf(minq, "%s", findfields)
		musts := []string{}
		for _, qs_wordsMust := range wordsMust {
			qws := strings.Split(qs_wordsMust, "+")
			mq := []string{}
			for _, qs_word := range qws {
				mq = append(mq, fmt.Sprintf(new_minq, qs_word))
			}
			musts = append(musts, fmt.Sprintf(NgramMust, strings.Join(mq, ",")))
		}
		querymust_string = strings.Join(musts, ",")
	}
	//log.Println("must", must, querymust_string)

	//querystring---------------------------------------------
	query_string := ""
	var querysShold []string
	if q, ok := querystring.(string); ok {
		if q != "" {
			querysShold = strings.Split(q, ",")
		}
	} else if q, ok := querystring.([]string); ok {
		querysShold = q
	} else if q, ok := querystring.([]interface{}); ok {
		querysShold = util.ObjArrToStringArr(q)
	}
	if querysShold != nil {
		for k, name := range strings.Split(findfields, ",") {
			for _, qs_querysShold := range querysShold {
				if k > 0 {
					query_string = query_string + "," + fmt.Sprintf(MatchQueryString, fmt.Sprint(name), qs_querysShold)
				} else {
					query_string = query_string + fmt.Sprintf(MatchQueryString, fmt.Sprint(name), qs_querysShold)
				}
			}
		}
	}
	//log.Println("querystring", query_string)
	if querymust_string == "" {
		qstring = fmt.Sprintf(FilterQuery_New, must, query_string, querymust_string)
	} else {
		qstring = fmt.Sprintf(FilterQuery_New, must, query_string, ","+querymust_string)
	}
	return
}
func GetByNgram(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int) *[]map[string]interface{} {
	return GetByNgramAll(index, itype, query, mustquery, findfields, order, fields, start, limit, false, false)
}

//增加高亮、过滤查询、高亮截取字数
func GetByNgramOther(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool, count int) *[]map[string]interface{} {
	defer util.Catch()
	qstr := ""
	if mustquery != "" && filtermode {
		qstr = GetNgramQuery(query, "", findfields)
		qstr = fmt.Sprintf(FilterQuery, mustquery, qstr[1:])
	} else {
		qstr = GetNgramQuery(query, mustquery, findfields)
	}
	if qstr != "" {
		if highlight {
			ws := []string{}
			for _, w := range strings.Split(findfields, ",") {
				ws = append(ws, fmt.Sprintf(highlightStr, w, count))
			}
			qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
		}
		if len(fields) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
		}
		if len(order) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
		}
		if start > -1 {
			qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
		}
		//log.Println("ngram-find", qstr)
		return Get(index, itype, qstr)
	} else {
		return nil
	}
}

//增加高亮、过滤查询
//替换了"号
func GetByNgramAll(index, itype string, query interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool) *[]map[string]interface{} {
	defer util.Catch()
	qstr := ""
	if mustquery != "" && filtermode {
		qstr = GetNgramQuery(query, "", findfields)
		qstr = fmt.Sprintf(FilterQuery, mustquery, qstr[1:])
	} else {
		qstr = GetNgramQuery(query, mustquery, findfields)
	}
	if qstr != "" {
		if highlight {
			ws := []string{}
			for _, w := range strings.Split(findfields, ",") {
				ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
			}
			qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
		}
		if len(fields) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
		}
		if len(order) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
		}
		if start > -1 {
			qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
		}
		//	log.Println("ngram-find", qstr)
		return Get(index, itype, qstr)
	} else {
		return nil
	}
}

//增加高亮、过滤查询
func GetByNgramAll_New(index, itype string, querystring, querymust interface{}, mustquery, findfields, order, fields string, start, limit int, highlight bool, filtermode bool) *[]map[string]interface{} {
	defer util.Catch()
	qstr := ""
	if filtermode {
		qstr = GetNgramQuery_New(querystring, querymust, mustquery, findfields)
	} else {
		qstr = GetNgramQuery_New(querystring, "", mustquery, findfields)
	}
	if qstr != "" {
		if highlight {
			ws := []string{}
			for _, w := range strings.Split(findfields, ",") {
				ws = append(ws, w+`:{"force_source": true}`)
			}
			qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL_New, strings.Join(ws, ",")) + `}`
		}
		if len(fields) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
		}
		if len(order) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", ",", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
		}
		if start > -1 {
			qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
		}
		//log.Println("ngram-find", order, qstr)
		return Get(index, itype, qstr)
	} else {
		return nil
	}
}

type KeyConfig struct {
	Keys      []string `json:"key"`
	NotKeys   []string `json:"notkey"`
	InfoTypes []string `json:"infotype"`
	Areas     []string `json:"area"`
}

//替换了"号
func GetResForJY(index, itype string, keys []KeyConfig, allquery, findfields, SortQuery, fields string, start, limit int) *[]map[string]interface{} {
	if len(keys) > 0 {
		qstr := ""
		new_minq := fmt.Sprintf(minq, "%s", findfields)
		not_new_minq := fmt.Sprintf(minq, "%s", `"title"`) //排除词只查询标题
		musts := []string{}
		for _, qs_words := range keys {
			mq := []string{}
			notmq := []string{}
			for _, qs_word := range qs_words.Keys {
				mq = append(mq, fmt.Sprintf(new_minq, ReplaceYH(qs_word)))
			}
			for _, qs_word := range qs_words.NotKeys {
				notmq = append(notmq, fmt.Sprintf(not_new_minq, ReplaceYH(qs_word)))
			}
			if len(qs_words.Areas) > 0 {
				mq = append(mq, fmt.Sprintf(`{"terms":{"area":["%s"]}}`, strings.Join(qs_words.Areas, `","`)))
			}
			if len(qs_words.InfoTypes) > 0 {
				mq = append(mq, fmt.Sprintf(`{"terms":{"toptype":["%s"]}}`, strings.Join(qs_words.InfoTypes, `","`)))
			}
			musts = append(musts, fmt.Sprintf(NgramMustAndNot, strings.Join(mq, ","), strings.Join(notmq, ",")))
		}
		qstr = fmt.Sprintf(NgramStr, "", strings.Join(musts, ","))

		qstr = fmt.Sprintf(FilterQuery, allquery, qstr[1:])
		ws := []string{}
		for _, w := range strings.Split(findfields, ",") {
			ws = append(ws, fmt.Sprintf(highlightStr, w, 1))
		}
		qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
		if len(fields) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
		}
		if len(SortQuery) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"sort":` + SortQuery + `}`
		}
		if start > -1 {
			qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
		}
		//log.Println("jy-ngram-find", qstr)
		return Get(index, itype, qstr)
	} else {
		return nil
	}
}

func ReplaceYH(src string) (rpl string) {
	return strings.Replace(src, `"`, `\"`, -1)
}

//
func GetAllByNgram(index, itype, qstr, findfields, order, fields string, start, limit, count int, highlight bool) *[]map[string]interface{} {
	if qstr != "" {
		if highlight {
			ws := []string{}
			for _, w := range strings.Split(findfields, ",") {
				ws = append(ws, fmt.Sprintf(highlightStr, w, count))
			}
			qstr = qstr[:len(qstr)-1] + `,` + fmt.Sprintf(HL, strings.Join(ws, ",")) + `}`
		}
		if len(fields) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"_source":[` + fields + "]}"
		}
		if len(order) > 0 {
			qstr = qstr[:len(qstr)-1] + `,"sort":[` + SR(SR(SR(SR(order, ",", "},{", -1), " ", "", -1), ":-1", `:"desc"`, -1), ":1", `:"asc"`, -1) + `]}`
		}
		if start > -1 {
			qstr = qstr[:len(qstr)-1] + `,"from":` + strconv.Itoa(start) + `,"size":` + strconv.Itoa(limit) + "}"
		}
		//log.Println("GetAllByNgram：", qstr)
		return Get(index, itype, qstr)
	} else {
		return nil
	}
}