generate answer

src/main/scala/com.gmei.up/Main.scala

@@ -30,8 +30,7 @@ class UserInfoDeserializationSchema extends KafkaDeserializationSchema[UserInfo]
 object Main {
   def main(args: Array[String]): Unit = {
     // println("###############")
-    // val query = ES.generateDiaryQuery(List("光子嫩肤", "水光针"), List("丰胸", "胸部塑身"), -1)
-    // query
+    // val query = ES.generateAnswerRequest(List("光子嫩肤", "水光针"), List("丰胸", "胸部塑身"))
     // println("###############")
 
     val env = StreamExecutionEnvironment.getExecutionEnvironment
@@ -57,12 +56,14 @@ object Main {
       val secondDemands = user.secondDemands.toList
       println(user.deviceId)
       println(projects.mkString(" "))
-      if ((projects.size > 0) || (secondDemands.size > 0)) {
-        val diaryQuery = ES.generateDiaryQuery(projects, secondDemands, -1)
-        val tractateQuery = ES.generateTractateQuery(projects, secondDemands)
-        ES.test(diaryQuery, tractateQuery)
-      }
+
+      val diaryReq = ES.generateDiaryRequest(projects, secondDemands, user.cityId)
+      val tractateReq = ES.generateTractateRequest(projects, secondDemands)
+      val answerReq = ES.generateAnswerRequest(projects, secondDemands)
+      ES.test(diaryReq, tractateReq, answerReq)
+
       println("########################")
+      user.deviceId
     }
 
     // stream.print

src/main/scala/com.gmei.up/utils/ES.scala

@@ -59,21 +59,23 @@ object ES {
     }
   }
 
-  def test(diaryQuery: SearchRequest, tractateQuery: String) = {
+  def test(diaryRequest: SearchRequest, tractateRequest: SearchRequest, answerRequest: SearchRequest) = {
     // TODO read from config
     val client: ElasticClient = ESClient.create("172.16.52.33", 9200, "elastic", "gengmei!@#")
 
-    val bb = multi(
-      diaryQuery,
-      search("gm-dbmw-tractate-read").source(tractateQuery)
+    val multiReq = multi(
+      diaryRequest,
+      tractateRequest,
+      answerRequest
     )
 
-    println(bb.show)
-    println(diaryQuery)
-    println(tractateQuery)
+    println(multiReq.show)
+    println(diaryRequest)
+    println(tractateRequest)
+    println(answerRequest)
 
     // TODO remove await
-    val resp = client.execute(bb).await
+    val resp = client.execute(multiReq).await
 
     // println(resp)
 
@@ -89,7 +91,7 @@ object ES {
     resp
   }
 
-  def generateDiaryQuery(
+  def generateDiaryRequest(
       projects: List[String],
       secondDemands: List[String],
       cityId: Int = -1
@@ -154,67 +156,101 @@ object ES {
     res
   }
 
-  def generateTractateQuery(projects: List[String], secondDemands: List[String]): String = {
-    val size = 30
+  def generateTractateRequest(projects: List[String], secondDemands: List[String]): SearchRequest = {
+    val size = 15
     val includes = List("id", "tags_v3")
+    var scoreListBuffer: ListBuffer[ScoreFunction] = ListBuffer.empty[ScoreFunction]
+    var shouldListBuffer: ListBuffer[Query] = ListBuffer.empty[Query]
 
-    val functions = new ListBuffer[JObject]()
-    val shoulds = new ListBuffer[JObject]()
     if (projects.size > 0) {
-      val promotes: List[JObject] =
-        List(("terms" -> ("tags_v3" -> projects)), ("term" -> ("is_promote" -> true)))
-
-      functions += ("filter" -> ("bool" -> ("must" -> promotes))) ~ ("weight" -> 60)
-
-      functions += ("filter" -> ("terms" -> { "tags_v3" -> projects })) ~ ("weight" -> 50)
-
-      shoulds += ("terms" -> ("tags_v3" -> projects))
+      scoreListBuffer += WeightScore(60).filter(
+        must(termsQuery("tags_v3", projects), termQuery("is_promote", true))
+      )
+      scoreListBuffer += WeightScore(50).filter(termsQuery("tags_v3", projects))
+      shouldListBuffer += TermsQuery("tags_v3", projects)
     }
-
     if (secondDemands.size > 0) {
-      functions += ("filter" -> ("terms" -> ("second_demands" -> secondDemands))) ~ ("weight" -> 40)
-
-      shoulds += ("terms" -> ("second_demands" -> secondDemands))
+      scoreListBuffer += WeightScore(40).filter(termsQuery("second_demands", secondDemands))
+      shouldListBuffer += TermsQuery("second_demands", secondDemands)
     }
 
-    val query = s"""
-{
-  "query": {
-    "function_score": {
-      "functions": ${pretty(render(functions.toList))},
-      "boost_mode": "replace",
-      "score_mode": "max",
-      "query": {
-        "bool": {
-          "should": ${pretty(render(shoulds.toList))},
-          "minimum_should_match": 1,
-          "must": [
-            {
-              "term": {
-                "is_online": "true"
+    val res = search("gm-dbmw-tractate-read")
+      .query {
+        functionScoreQuery()
+          .functions(scoreListBuffer.toList)
+          .scoreMode(FunctionScoreQueryScoreMode.Max)
+          .boostMode(CombineFunction.Replace)
+          .query {
+            boolQuery()
+              .must(
+                termQuery("is_online", true),
+                rangeQuery("content_level").gte(3)
+              )
+              .not(termQuery("show_by_index", 2), termQuery("status", 4))
+              .should(shouldListBuffer.toList)
+              .minimumShouldMatch(1)
           }
-            },
-            {
-              "terms": {
-                "content_level": [6, 5, 4, 3.5, 3]
       }
+      .size(size)
+      .sourceInclude(includes)
+      .sortBy(
+        FieldSort("_score").desc()
+      )
+
+    // println("tractate query")
+    // println(res.show)
+
+    res
   }
-          ],
-          "must_not": [
-            { "term": { "status": 4 } },
-            { "term": { "show_by_index": 2 } }
-          ]
+
+  def generateAnswerRequest(projects: List[String], secondDemands: List[String]): SearchRequest = {
+    val size = 15
+    val includes = List("id", "tags_v3")
+    var scoreListBuffer: ListBuffer[ScoreFunction] = ListBuffer.empty[ScoreFunction]
+    var shouldListBuffer: ListBuffer[Query] = ListBuffer.empty[Query]
+
+    if (projects.size > 0) {
+      scoreListBuffer += WeightScore(60).filter(
+        must(termsQuery("tags_v3", projects), termQuery("is_promote", true))
+      )
+      scoreListBuffer += WeightScore(50).filter(termsQuery("tags_v3", projects))
+      shouldListBuffer += TermsQuery("tags_v3", projects)
     }
+    if (secondDemands.size > 0) {
+      scoreListBuffer += WeightScore(40).filter(termsQuery("second_demands", secondDemands))
+      shouldListBuffer += TermsQuery("second_demands", secondDemands)
+    }
+
+    val res = search("gm-dbmw-tractate-read")
+      .query {
+        functionScoreQuery()
+          .functions(scoreListBuffer.toList)
+          .scoreMode(FunctionScoreQueryScoreMode.Max)
+          .boostMode(CombineFunction.Replace)
+          .query {
+            boolQuery()
+              .must(
+                termQuery("is_online", true),
+                rangeQuery("content_level").gte(3),
+                rangeQuery("content_length").gte(30)
+              )
+              .not(termQuery("show_by_index", 2))
+              .should(shouldListBuffer.toList)
+              .minimumShouldMatch(1)
           }
       }
-  },
-  "size": ${size},
-  "_source": {"include": ${pretty(render(includes))}}
-}
-""".stripMargin
+      .size(size)
+      .sourceInclude(includes)
+      .sortBy(
+        FieldSort("_score").desc(),
+        FieldSort("has_picture").desc(),
+        FieldSort("smart_rank_v2").desc(),
+        FieldSort("good_click").desc()
+      )
 
-    // println("tractate query:")
-    // println(query)
-    query
+    // println("answer query")
+    // println(res.show)
+
+    res
   }
 }

src/main/scala/com.gmei.up/utils/User.scala

@@ -5,6 +5,8 @@ case class UserInfo(
     action: String,
     logTime: Double,
     eventCn: String,
+    cityId: Int,
     secondDemands: Array[String],
-    projects: Array[String]
+    projects: Array[String],
+    bussinessTags: Array[String]
 )