<commit_before>package com.microsoft.partnercatalyst.fortis.spark.pipeline

import com.github.catalystcode.fortis.spark.streaming.instagram.dto.InstagramItem
import com.microsoft.partnercatalyst.fortis.spark.dto.AnalyzedItem
import com.microsoft.partnercatalyst.fortis.spark.streamprovider.{ConnectorConfig, StreamProvider}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

object InstagramPipeline extends Pipeline {
  override def apply(streamProvider: StreamProvider, streamRegistry: Map[String, List[ConnectorConfig]],
    ssc: StreamingContext, transformContext: TransformContext): Option[DStream[AnalyzedItem]] = {
    import transformContext._

    streamProvider.buildStream[InstagramItem](ssc, streamRegistry("instagram")).map(_
      .map(instagram => {
        // do computer vision analysis
        val analysis = imageAnalyzer.analyze(instagram.images.standard_resolution.url)
        AnalyzedItem(
          body = instagram.caption.text,
          title = "",
          sharedLocations = instagram.location match {
            case Some(location) => locationsExtractor.fetch(location.latitude, location.longitude).toList
            case None => List()},
          analysis = analysis.copy(
            keywords = analysis.keywords.filter(tag => targetKeywords.contains(tag.name))),
          source = instagram.link)
      })
      .map(analyzedItem => {
        // keyword extraction
        val keywords = keywordExtractor.extractKeywords(analyzedItem.body)
        analyzedItem.copy(analysis = analyzedItem.analysis.copy(keywords = keywords ::: analyzedItem.analysis.keywords))
      })
      .filter(_.analysis.keywords.nonEmpty))
  }
}<commit_msg>Consider image description in keyword generation<commit_after>package com.microsoft.partnercatalyst.fortis.spark.pipeline

import com.github.catalystcode.fortis.spark.streaming.instagram.dto.InstagramItem
import com.microsoft.partnercatalyst.fortis.spark.dto.AnalyzedItem
import com.microsoft.partnercatalyst.fortis.spark.streamprovider.{ConnectorConfig, StreamProvider}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

object InstagramPipeline extends Pipeline {
  override def apply(streamProvider: StreamProvider, streamRegistry: Map[String, List[ConnectorConfig]],
    ssc: StreamingContext, transformContext: TransformContext): Option[DStream[AnalyzedItem]] = {
    import transformContext._

    streamProvider.buildStream[InstagramItem](ssc, streamRegistry("instagram")).map(_
      .map(instagram => {
        // do computer vision analysis
        val analysis = imageAnalyzer.analyze(instagram.images.standard_resolution.url)
        AnalyzedItem(
          body = analysis.summary.getOrElse(""),
          title = instagram.caption.text,
          sharedLocations = instagram.location match {
            case Some(location) => locationsExtractor.fetch(location.latitude, location.longitude).toList
            case None => List()},
          analysis = analysis.copy(
            keywords = analysis.keywords.filter(tag => targetKeywords.contains(tag.name))),
          source = instagram.link)
      })
      .map(analyzedItem => {
        // keyword extraction
        val keywords = keywordExtractor.extractKeywords(analyzedItem.title) ::: keywordExtractor.extractKeywords(analyzedItem.body)
        analyzedItem.copy(analysis = analyzedItem.analysis.copy(keywords = keywords ::: analyzedItem.analysis.keywords))
      })
      .filter(_.analysis.keywords.nonEmpty))
  }
}
{'fraction_non_alphanumeric': 0.10196629213483147, 'fraction_numerical': 0.0, 'mean_word_length': 3.9321329639889195, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7064150', 'n_tokens_mistral': 980, 'n_tokens_neox': 940, 'n_words': 153}