es.py

# -*- coding: utf-8 -*-

from tornado.options import options
from pyes import (
    ES,
    Search,
    FilteredQuery,
    ANDFilter,
    TermFilter,
    ORFilter,
    MatchAllQuery,
    MatchQuery,
    BoolQuery,
    TermsQuery,
    TermQuery,
    ConstantScoreQuery,
)

from luffy.utils.convert import MACRO, pagination

__all__ = ['ESDAO']

class ESDAO(object):
    @classmethod
    def get_conn(self):
        return ES(options.es_host)

    @classmethod
    def search_job(self, keyword, function_id=None, region_id=None, job_type=None, page=1, size=MACRO.DEFAULT_COUNT, sort='created_at:desc'):
        conn = self.get_conn()
        filters = [TermFilter('is_deleted', 0)]
        if keyword:
            keyword = keyword.lower()
            query = BoolQuery(
                should = [ ConstantScoreQuery(MatchQuery('company_name', keyword, operator='and'), boost=1000)
                         , ConstantScoreQuery(MatchQuery('company_full_name', keyword, operator='and'), boost=800)
                         , ConstantScoreQuery(MatchQuery('title', keyword, operator='and'), boost=500)
                         #, ConstantScoreQuery(MatchQuery('description', keyword, operator='and'), boost=10)
                         #, ConstantScoreQuery(MatchQuery('company_intro', keyword, operator='and'), boost=10)
                         ])
            # query = MatchQuery('company_name', keyword)#, type='phrase')
        else:
            query = MatchAllQuery()
        if function_id:
            filters.append(TermFilter('function_id', function_id))
        if region_id:
            filters.append(ORFilter([ TermFilter('region_id', region_id)
                                      , TermFilter('top_region_id', region_id)
                                    ]))
        if job_type:
            filters.append(TermFilter('job_type', job_type))
        if filters:
            f = ANDFilter(filters)
            q = FilteredQuery(query, f).search()
        else:
            q = query.search()
        result = conn.search(q, 'truffle', 'job', sort=sort)
        # result = conn.search(q, 'truffle', 'job')
        return pagination(page, size, result)

es.rb

module NSB::API::Helpers::Search
  class ES_Client
    def initialize
      host = Application.config.elasticsearch_host
      port = Application.config.elasticsearch_port
      index = Application.config.elasticsearch_index
      @conn = Stretcher::Server.new("http://#{host}:#{port}").index(index.to_sym).type(:job)
      @filters = {and: []}
      @query = {match_all: {}}
    end

    def parse_status
      @filters[:and].push({term: {is_deleted: 0}})
    end

    def parse_region(region_id)
      @filters[:and].push({or: [{term: {region_id: region_id}}, {term: {top_region_id: region_id}}]})
    end

    def parse_function(function_id)
      @filters[:and].push({term: {function_id: function_id}})
    end

    def parse_type(type_id)
      @filters[:and].push({term: {job_type: type_id}})
    end

    def parse_timestamp(time_range)
      {
        range: {
          created_at: time_range || {}
        }
      }
    end

    def parse_q(keyword)
      keyword = keyword.strip.downcase
      q_query = [ {field_name: :company_name, boost: 1000},
                  {field_name: :company_full_name, boost: 800},
                  {field_name: :title, boost: 500},
                  # {field_name: :description, boost: 10},
                  # {field_name: :company_intro, boost: 10},
                ]
      q_query.map! { |i|
        { constant_score: {
            query: {
              match: {
                i[:field_name] => {
                  operator: "and",
                  query: keyword,
                  type: "boolean",
                }
              }
            },
            boost: i[:boost]
          }
        }
      }
    end

    def search(params)
      parse_status
      parse_region(params[:region_id]) if params[:region_id]
      parse_function(params[:function_id]) if params[:function_id]
      parse_type(params[:type_id])if params[:type_id]
      q_query = parse_q(params[:q]) if params[:q]

      @query = {
        bool: {
          must: parse_timestamp(params[:time_range]),
          minimum_number_should_match: 1,
          should: q_query,
        }
      }

      @conn.search({
        query: {
          filtered: {
          filter: @filters,
          query: @query,
          }
        },
        from: params[:offset],
        fields: [:_id],
        size: params[:limit],
        # Filter outside
        sort: [created_at: "desc"],
      })
    end
  end

  def search(params)
    es = ES_Client.new
    es.search(params)
  end
end