<commit_before>package com.soteradefense.datawake.trails.topology.search.bolts

import java.io.InputStreamReader
import java.net.{URL, URLEncoder}
import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util

import backtype.storm.task.TopologyContext
import backtype.storm.topology.base.BaseBasicBolt
import backtype.storm.topology.{BasicOutputCollector, OutputFieldsDeclarer}
import backtype.storm.tuple.{Fields, Tuple, Values}
import com.soteradefense.datawake.trails.constants.DatawakeConstants
import com.soteradefense.datawake.trails.rank.DatawakeUrlRankHelper
import com.soteradefense.datawake.trails.sql.SqlCredentials
import com.soteradefense.datawake.trails.topology.search.data.{DatawakeInternetResult, DatawakeSearchResults}
import com.soteradefense.datawake.trails.topology.search.json.{GoogleResults, Result, SearchResults}
import net.ettinsmoor.{Bingerator, WebResult}
import net.liftweb.json._
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

class SearchBolt(sqlCredentials: SqlCredentials, newUrl: Fields, newTerm: Fields, selectPageRank: String, resultUpdateSql: String, invalidResultUpdateSql: String, updatePageRankSql: String) extends BaseBasicBolt {

  var GOOGLE_CX: String = null
  var GOOGLE_KEY: String = null
  var GOOGLE_API: String = null
  val DEFAULT_CHARSET: String = "UTF-8"
  val RESULT_SET: String = "&rsz=8"
  var connection: Connection = null
  var bingKey: String = null
  var logger: Logger = null
  var SEARCH_API: String = null

  override def prepare(stormConf: util.Map[_, _], context: TopologyContext): Unit = {
    super.prepare(stormConf, context)
    Class.forName("com.mysql.jdbc.Driver")
    connection = DriverManager.getConnection(sqlCredentials.jdbc, sqlCredentials.username, sqlCredentials.password)
    bingKey = stormConf.get(DatawakeConstants.BING_KEY_ID).asInstanceOf[String]
    GOOGLE_KEY = stormConf.get(DatawakeConstants.GOOGLE_KEY_ID).asInstanceOf[String]
    GOOGLE_CX = stormConf.get(DatawakeConstants.GOOGLE_CX_ID).asInstanceOf[String]
    SEARCH_API = stormConf.get(DatawakeConstants.SEARCH_ENDPOINT_ID).asInstanceOf[String]
    logger = LoggerFactory.getLogger(this.getClass)

    GOOGLE_API = "https://www.googleapis.com/customsearch/v1?key=" + URLEncoder.encode(GOOGLE_KEY, DEFAULT_CHARSET) + "&cx=" + URLEncoder.encode(GOOGLE_CX, DEFAULT_CHARSET) + "&num=10&q="
  }

  override def execute(input: Tuple, collector: BasicOutputCollector): Unit = {
    val searchTerm: String = input.getStringByField("kafkaTerm")
    val org: String = input.getStringByField("kafkaOrg")
    val domain: String = input.getStringByField("kafkaDomain")
    val trail: String = input.getStringByField("kafkaTrail")
    val isRelevantTerm: Boolean = input.getBooleanByField("kafkaValid")
    val customSearch: GoogleResults = getGoogleResultsFromServer(searchTerm, 0)
    if (customSearch != null) {
      if (isRelevantTerm) {
        logger.info("Total Result Count For {}: {}", searchTerm, customSearch.results.size)
        var rank: Int = 0
        customSearch.results.foreach(f => {
          var pageRank = DatawakeUrlRankHelper.getPageRank(org, domain, trail, f.url, selectPageRank, connection)
          if (pageRank == -1) {
            collector.emit("new-url", new Values(org, domain, trail, f.url, f.title, rank.asInstanceOf[Integer]))
          } else {
            pageRank = Math.min(pageRank, rank)
            updatePageRank(org, domain, trail, f.url, pageRank)
          }

          rank = rank + 1
        })
        logger.info("Adding relevant entity: {} with {} pages ", searchTerm, customSearch.results.size)
        updateResultCount(resultUpdateSql, org, domain, trail, searchTerm, customSearch.results.size)
      } else {
        updateResultCount(invalidResultUpdateSql, org, domain, trail, searchTerm, customSearch.resultCount)
      }
    }

    collector.emit("new-term", new Values(org, domain, trail))
  }

  def updatePageRank(org: String, domain: String, trail: String, url: String, rank: Int) = {
    var countPrepare: PreparedStatement = null
    try {<commit_msg>changing size field being stored in db<commit_after>package com.soteradefense.datawake.trails.topology.search.bolts

import java.io.InputStreamReader
import java.net.{URL, URLEncoder}
import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util

import backtype.storm.task.TopologyContext
import backtype.storm.topology.base.BaseBasicBolt
import backtype.storm.topology.{BasicOutputCollector, OutputFieldsDeclarer}
import backtype.storm.tuple.{Fields, Tuple, Values}
import com.soteradefense.datawake.trails.constants.DatawakeConstants
import com.soteradefense.datawake.trails.rank.DatawakeUrlRankHelper
import com.soteradefense.datawake.trails.sql.SqlCredentials
import com.soteradefense.datawake.trails.topology.search.data.{DatawakeInternetResult, DatawakeSearchResults}
import com.soteradefense.datawake.trails.topology.search.json.{GoogleResults, Result, SearchResults}
import net.ettinsmoor.{Bingerator, WebResult}
import net.liftweb.json._
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

class SearchBolt(sqlCredentials: SqlCredentials, newUrl: Fields, newTerm: Fields, selectPageRank: String, resultUpdateSql: String, invalidResultUpdateSql: String, updatePageRankSql: String) extends BaseBasicBolt {

  var GOOGLE_CX: String = null
  var GOOGLE_KEY: String = null
  var GOOGLE_API: String = null
  val DEFAULT_CHARSET: String = "UTF-8"
  val RESULT_SET: String = "&rsz=8"
  var connection: Connection = null
  var bingKey: String = null
  var logger: Logger = null
  var SEARCH_API: String = null

  override def prepare(stormConf: util.Map[_, _], context: TopologyContext): Unit = {
    super.prepare(stormConf, context)
    Class.forName("com.mysql.jdbc.Driver")
    connection = DriverManager.getConnection(sqlCredentials.jdbc, sqlCredentials.username, sqlCredentials.password)
    bingKey = stormConf.get(DatawakeConstants.BING_KEY_ID).asInstanceOf[String]
    GOOGLE_KEY = stormConf.get(DatawakeConstants.GOOGLE_KEY_ID).asInstanceOf[String]
    GOOGLE_CX = stormConf.get(DatawakeConstants.GOOGLE_CX_ID).asInstanceOf[String]
    SEARCH_API = stormConf.get(DatawakeConstants.SEARCH_ENDPOINT_ID).asInstanceOf[String]
    logger = LoggerFactory.getLogger(this.getClass)

    GOOGLE_API = "https://www.googleapis.com/customsearch/v1?key=" + URLEncoder.encode(GOOGLE_KEY, DEFAULT_CHARSET) + "&cx=" + URLEncoder.encode(GOOGLE_CX, DEFAULT_CHARSET) + "&num=10&q="
  }

  override def execute(input: Tuple, collector: BasicOutputCollector): Unit = {
    val searchTerm: String = input.getStringByField("kafkaTerm")
    val org: String = input.getStringByField("kafkaOrg")
    val domain: String = input.getStringByField("kafkaDomain")
    val trail: String = input.getStringByField("kafkaTrail")
    val isRelevantTerm: Boolean = input.getBooleanByField("kafkaValid")
    val customSearch: GoogleResults = getGoogleResultsFromServer(searchTerm, 0)
    if (customSearch != null) {
      if (isRelevantTerm) {
        logger.info("Total Result Count For {}: {}", searchTerm, customSearch.results.size)
        logger.inof("Google results: {}", customSearch)
        var rank: Int = 0
        customSearch.results.foreach(f => {
          var pageRank = DatawakeUrlRankHelper.getPageRank(org, domain, trail, f.url, selectPageRank, connection)
          if (pageRank == -1) {
            collector.emit("new-url", new Values(org, domain, trail, f.url, f.title, rank.asInstanceOf[Integer]))
          } else {
            pageRank = Math.min(pageRank, rank)
            updatePageRank(org, domain, trail, f.url, pageRank)
          }

          rank = rank + 1
        })
        logger.info("Adding relevant entity: {} with {} pages ", searchTerm, customSearch.results.size)
        updateResultCount(resultUpdateSql, org, domain, trail, searchTerm, customSearch.results.size.toString)
      } else {
        updateResultCount(invalidResultUpdateSql, org, domain, trail, searchTerm, customSearch.resultCount)
      }
    }

    collector.emit("new-term", new Values(org, domain, trail))
  }

  def updatePageRank(org: String, domain: String, trail: String, url: String, rank: Int) = {
    var countPrepare: PreparedStatement = null
    try {
{'fraction_non_alphanumeric': 0.1058542413381123, 'fraction_numerical': 0.0023894862604540022, 'mean_word_length': 4.706203135650989, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6102472', 'n_tokens_mistral': 2617, 'n_tokens_neox': 2573, 'n_words': 517}