background/model/port.go

package model

import (
	"background/db"
	"reflect"
	"strings"
	"ubntgo/logger"

	"github.com/go-openapi/errors"
	json "github.com/json-iterator/go"
	"gopkg.in/olivere/elastic.v3"
	"qiniupkg.com/x/log.v7"
)

type Field struct {
	Field string  `sql:"Field"`
	Type string  `sql:"Type"`
	Key string `sql:"Key"`
}


func MysqlToElasticSearchMapping(types string,Key string) string{
	if Key == "PRI"{
		return "keyword"
	}
	if(strings.Contains(types,"int(")){
		return "integer"
	}
	if(strings.Contains(types,"longblob")){
		return "text"
	}
	if(strings.Contains(types,"varchar")){
		return "text"
	}
	if(strings.Contains(types,"datetime")){
		return "date"
	}
	return ""
}
/*
"settings":{
		"number_of_shards":1,
		"number_of_replicas":0
	},
	"mappings":{
		"properties":{
			"user":{
				"type":"keyword"
			},
			"message":{
				"type":"text",
				"store": true,
				"fielddata": true
			},
			"tags":{
				"type":"keyword"
			},
			"location":{
				"type":"geo_point"
			},
			"suggest_field":{
				"type":"completion"
			}
		}
	}
*/
// 不同类型db之间进行缓存
func QueryDocument(query elastic.Query,limit int,offset int) ([]Hardware,error){
	var ret []Hardware
	data,ids,e := db.GetElastic().Query("doc",query,reflect.TypeOf(Hardware{}),limit,offset)
	log.Print(data)
	if nil != e{
		return nil,e
	}
	i := 0
	for _,v := range data{
		ret = append(ret,v.(Hardware))
		ret[i].ID = ids[i]
		i++
	}
	return ret,nil
}
func InsertDocToElaticSearch(doc Doc)  error{

	matchPhraseQuery := elastic.NewMatchQuery("title", doc.Title)
	existedHardware,e := QueryHardwares(matchPhraseQuery,10,0)
	log.Print(e,existedHardware)

	for _,v := range existedHardware{
		if v.Name == doc.Title{
			log.Print(v.ID)
			return errors.New(200,"existed title")
		}
	}
	e = db.GetElastic().Create("doc","0","",doc)
	if nil != e{
		log.Print(e.Error())
		return e
	}
	return nil
}

func PortDocumentToElasticsearch(tblname string )  error{
	ret,e := GetAllDocs()
	if nil != e{
		log.Print(e.Error())
	}
	insert,err := json.Marshal(ret)
	if nil != err{
		log.Print(err)
	}
	log.Print(len(ret))
	log.Print(string(insert))

	columns := []Field{}
	e = db.GetMysqlClient().Query2("describe " + tblname,&columns)
	if nil != e{
		logger.Debug(e.Error())
		return e
	}
	if existed,_ := db.GetElastic().IndexExisted(tblname);existed{
		for _,v := range ret{
			e := InsertDocToElaticSearch(v)
			if nil != e{
				log.Print(e.Error())
			}
			log.Print("index not existed , create " + tblname)
		}
	}else{
		props := map[string]interface{}{}
		mapping := map[string]interface{}{
			"settings":map[string]interface{}{
				"analysis":map[string]interface{}{
					"analyzer":map[string]interface{}{
						"default":map[string]interface{}{
							"type": "smartcn",
						},
					},
				},
			},
			"mappings": map[string]interface{}{
				"properties":props,
			},
		}

		for _,v := range columns{
			props[v.Field] = map[string]string{"type":MysqlToElasticSearchMapping(v.Type,v.Key)};
		}
		dat,e := json.Marshal(mapping)
		if nil != e{
			log.Print(e.Error())
		}
		e = db.GetElastic().CreateIndex(tblname,string(dat))
		if nil != e{
			log.Print(e.Error())
		}
		log.Print(string(dat))
		for _,v := range ret{
			e := InsertDocToElaticSearch(v)
			if nil != e{
				log.Print(e.Error())
			}
		}
	}

	return nil
}