MediaHub Application: search files

README.md

@@ -1,29 +1,22 @@
 # mediahub
 
+Bilan: 
+<br>Voici ma Solution a l"exercice de recherche d'information dans des fichiers TSV en utilisant la bibliotheque Akka Stream realiser en Java et Scala
+<br>Le service de recherche est fait en Scala, celui-ci est intègrer dans une solution Springboot en Java
 
-Instruction :
-Vous allez avoir droit à 5 fichier.
+Instruction:
+<br>Pour le bon fonctionement du projet, il est necessaire de telecharger les fichiers suivant et de les placer dans le repertoire resources, ne pas dezipper les fichiers (main\resources): 
 
-Ces 5 fichiers Charles pense qu'il faudra les charger dans dictionnary, dans 5 map et ensuite vous allez devoir les requêté à l’aide d’akkastream.
+- https://datasets.imdbws.com/name.basics.tsv.gz
+- https://datasets.imdbws.com/title.episode.tsv.gz
+- https://datasets.imdbws.com/title.ratings.tsv.gz
+- https://datasets.imdbws.com/title.principals.tsv.gz
+- https://datasets.imdbws.com/title.basics.tsv.gz
 
-Le principe d’akka est que sa permet de créer un pipe, vous avez la même logique que des streams en java. Donc le principe est que vous avez deux, trois recherche à opérer et vous allez faire un cor ( une api business) qui permet de faire une recherche et qu’il faudra écrire en SCALA.
+Afin de lancer la solution, il suffit de lancer MediahubApplication qui se trouve dans la partie Java du projet
 
-L’idéal serait d’écrire tous le service en scala, regarder sur internet comment créer un service en scala.
+Les deux endpoints disponibles pour tester le service sont:
 
-Sinon,  vous faites un springboot qui va appeler ce core dev en scala. Ce qui est certain c’est que le core doit être écrit en scala sur la base de l’api stream AKKA.
+- http://localhost:8080/api/teamMembers?movieTitle=Carmencita
+- http://localhost:8080/api/tvSeries
 
-Le sentiment de Charles est qu’il faut bien montrer le principe du pipe, c’est à dire que vous allez avoir potentiellement 5 map et l’idée est d’avoir un seul stream qui va permettre de les requêtes et de concaténer les  résultats, pour en faire un résultat final qui va être envoyé au service.
-
-Vous en discuterez lundi avec Charles pour savoir de quelle façon le faire.
-
-
-
-Pour résumer :
-
-Un core qui va être développé en scala au travers de la librairie Akkastream ( akkastream va permettre de requêter les différend fichier comme étant des source de donnée).
-
-Sur la base de cette source-là vous allez avoir des sources de données que vous allez concaténer au travers d’un stream et c’est sur ce stream là que vous allez faire vos recherches.
-
-A voir comment vous allez gérer ça lundi avec Charles.
-
-Ce service, il faudra l’exposer avec un springboot qui prépose juste une API requête qui appellera ce cœur business.

pom.xml

@@ -25,6 +25,30 @@
     <build>
         <plugins>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.8.1</version>
+                <configuration>
+                    <source>18</source>
+                    <target>18</target>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.2.4</version>
+                <configuration>
+                </configuration>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
                 <groupId>net.alchim31.maven</groupId>
                 <artifactId>scala-maven-plugin</artifactId>
                 <version>3.1.3</version>
@@ -64,6 +88,19 @@
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-stream_${scala.binary.version}</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.lightbend.akka</groupId>
+            <artifactId>akka-stream-alpakka-csv_${scala.binary.version}</artifactId>
+            <version>4.0.0</version>
+        </dependency>
+
+        <!-- alpakka -->
+        <dependency>
+            <groupId>com.lightbend.akka</groupId>
+            <artifactId>akka-stream-alpakka-file_2.13</artifactId>
+            <version>4.0.0</version>
+        </dependency>
+
         <!-- SpringBoot -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
@@ -85,6 +122,27 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.module/jackson-module-scala -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.module</groupId>
+            <artifactId>jackson-module-scala_2.13</artifactId>
+            <version>2.14.0-rc1</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.json4s/json4s-native -->
+        <dependency>
+            <groupId>org.json4s</groupId>
+            <artifactId>json4s-native_2.13</artifactId>
+            <version>4.1.0-M1</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.json4s/json4s -->
+        <dependency>
+            <groupId>org.json4s</groupId>
+            <artifactId>json4s_2.11</artifactId>
+            <version>3.2.11</version>
+        </dependency>
+
     </dependencies>
 
+
+
 </project>

src/main/java/com/mediahub/MediahubApplication.java

@@ -0,0 +1,13 @@
+package com.mediahub;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class MediahubApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(MediahubApplication.class, args);
+    }
+
+}

src/main/java/com/mediahub/controller/MovieController.java

@@ -0,0 +1,53 @@
+package com.mediahub.controller;
+
+import com.mediahub.JsonCustomMapping;
+import com.mediahub.MovieService;
+import com.mediahub.MovieServiceTrait;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+
+@RestController
+@RequestMapping("/api")
+public class MovieController {
+    @Autowired
+    MovieServiceTrait movieServiceTrait;
+
+    @GetMapping("/teamMembers")
+    public ResponseEntity getTeamMembers(@RequestParam(required = true) String movieTitle) {
+        scala.collection.immutable.List<MovieService.Principal> principalList;
+
+        try {
+            if (movieTitle == null)
+                return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
+            else {
+                principalList = (movieServiceTrait.principalsForMovieName(movieTitle).toList());
+            }
+
+            return new ResponseEntity<>(new JsonCustomMapping().mapperPrincipal(principalList), HttpStatus.OK);
+
+        } catch (Exception e) {
+            return new ResponseEntity<>(null, HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+
+    }
+
+    @GetMapping("/tvSeries")
+    public ResponseEntity getTvSeries() {
+        scala.collection.immutable.List<MovieService.TvSeries> tvSeriesList;
+
+        try {
+            tvSeriesList = (movieServiceTrait.tvSeriesWithGreatestNumberOfEpisodes().toList());
+            return new ResponseEntity<>(new JsonCustomMapping().mapperSeries(tvSeriesList), HttpStatus.OK);
+        } catch (Exception e) {
+            return new ResponseEntity<>(null, HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+
+    }
+
+}

src/main/scala/com/mediahub/JsonCustomMapping.scala

@@ -0,0 +1,15 @@
+package com.mediahub
+
+import org.json4s.DefaultFormats
+import org.json4s.native.Json
+
+class JsonCustomMapping extends MovieService {
+
+  def mapperPrincipal(list:  List[Principal]): String = {
+     Json(DefaultFormats).write(list)
+  }
+
+  def mapperSeries(list: List[TvSeries]): String = {
+    Json(DefaultFormats).write(list)
+  }
+}

src/main/scala/com/mediahub/MovieQueryService.scala

@@ -0,0 +1,146 @@
+package com.mediahub
+
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.stream.ActorAttributes.supervisionStrategy
+import akka.stream.Supervision.resumingDecider
+import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap}
+import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source}
+import akka.util.ByteString
+import org.springframework.stereotype.Component
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Paths
+import scala.collection.{immutable, mutable}
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
+import scala.language.postfixOps
+import scala.util.Try
+
+
+@Component
+class MovieQueryService extends MovieServiceTrait {
+
+  implicit val system: ActorSystem = ActorSystem()
+
+  val titleBasicsResource: File = new File("src/main/resources/title.basics.tsv.gz")
+  val titlePrincipalsResource: File = new File("src/main/resources/title.principals.tsv.gz")
+  val nameBasicsResource: File = new File("src/main/resources/name.basics.tsv.gz")
+  val titleEpisodeResource: File = new File("src/main/resources/title.episode.tsv.gz")
+
+  def fileSource(file: File): Source[ByteString, NotUsed] = {
+    Source.single(file)
+      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024 * 2))
+      .via(Compression.gunzip())
+  }
+
+  val lineParser: Flow[ByteString, Map[String, String], NotUsed] = {
+    CsvParsing
+      .lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote)
+      .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
+  }
+
+  override def principalsForMovieName(name: String): List[Principal] = {
+
+    val titleId: String = getIdOfTitle(titleBasicsResource, name)
+    val personsIdList: List[String] = getIdOfPersons(titlePrincipalsResource, titleId).toList.flatten
+    val personsList = getPersons(nameBasicsResource, personsIdList)
+    personsList
+  }
+
+  override def tvSeriesWithGreatestNumberOfEpisodes(): List[TvSeries] = {
+    val seriesTitleMap = getSeriesTitle(titleEpisodeResource)
+    val titleListMaxEpisode = getTitleByIds(titleBasicsResource, seriesTitleMap)
+    titleListMaxEpisode
+  }
+
+
+
+  def getIdOfTitle(file: File, titleName: String): String = {
+    val result = fileSource(file)
+      .mapAsync(50) {
+        res =>
+          Source.single(res)
+            .via(lineParser)
+            .filter(row => row.getOrElse("primaryTitle", "") == titleName)
+            .map(a => a.get("tconst"))
+            .withAttributes(supervisionStrategy(resumingDecider))
+            .runWith(Sink.head)
+      }
+      .withAttributes(supervisionStrategy(resumingDecider))
+      .runWith(Sink.head)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.get
+  }
+
+  def getTitleByIds(file: File, titleIdMap: mutable.Map[String, Int]): List[TvSeries] = {
+    val result = Source.single(file)
+      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+      .via(Compression.gunzip())
+      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
+      .via(CsvToMap.toMapAsStringsCombineAll(StandardCharsets.UTF_8, Option.empty))
+      .filter(row => titleIdMap.keySet.toList.contains(row.getOrElse("tconst", "")))
+      .map(a => TvSeries(a("originalTitle"), Try(a.getOrElse("startYear","").toInt).getOrElse(0), a.getOrElse("endYear","").toIntOption, a.get("genres").toList))
+      .runWith(Sink.collection)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.toList
+  }
+
+  def getIdOfPersons(file: File, titleId: String): immutable.Iterable[Option[String]] = {
+    val result = fileSource(file)
+      .mapAsync(50) {
+        res =>
+          Source.single(res)
+            .via(lineParser)
+            .filter(row => row.getOrElse("tconst", "") == titleId)
+            .map(a => a.get("nconst"))
+            .withAttributes(supervisionStrategy(resumingDecider))
+            .runWith(Sink.collection)
+      }
+      .withAttributes(supervisionStrategy(resumingDecider))
+      .runWith(Sink.head)
+
+    Await.result(result, 5 minutes)
+  }
+
+
+  def getPersons(file: File, personsIdList: List[String]): List[Principal] = {
+    val result = Source.single(file)
+      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+      .via(Compression.gunzip())
+      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
+      .via(CsvToMap.toMapAsStringsCombineAll(StandardCharsets.UTF_8, Option.empty))
+      .filter(row => personsIdList.contains(row.getOrElse("nconst", "")))
+      .map(a => Principal(a("primaryName"), Try(a("birthYear").toInt).getOrElse(0), a("deathYear").toIntOption, a.get("primaryProfession").toList))
+      .runWith(Sink.collection)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.toList
+  }
+
+  def getSeriesTitle(file: File): mutable.Map[String, Int] = {
+
+    val seriesMap = collection.mutable.Map("0" -> 0)
+    val maxSeriesMap = collection.mutable.Map("0" -> 0)
+
+    val result = Source.single(file)
+      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+      .via(Compression.gunzip())
+      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
+      .via(CsvToMap.toMapAsStringsCombineAll(StandardCharsets.UTF_8, Option.empty))
+      .map(row => seriesMap += seriesMap.get(row("parentTconst")).map(x => row("parentTconst") -> (x + 1)).getOrElse(row("parentTconst") -> 1))
+      .runWith(Sink.seq)
+
+    Await.result(result, 5 minutes)
+
+    for (_ <- 1 to 10) {
+      maxSeriesMap.addOne(seriesMap.maxBy(x => x._2)._1 -> seriesMap.maxBy(x => x._2)._2)
+      seriesMap.remove(seriesMap.maxBy(x => x._2)._1)
+    }
+    maxSeriesMap
+  }
+
+}

src/main/scala/com/mediahub/MovieService.scala

@@ -0,0 +1,20 @@
+package com.mediahub
+
+class MovieService {
+
+  final case class Principal(
+                              name: String,
+                              birthYear: Int,
+                              deathYear: Option[Int],
+                              profession: List[String])
+
+  case class TvSeries(
+                       original: String,
+                       startYear: Int,
+                       endYear: Option[Int],
+                       genres: List[String])
+
+
+}
+
+

src/main/scala/com/mediahub/MovieServiceTrait.scala

@@ -0,0 +1,8 @@
+package com.mediahub
+
+
+trait MovieServiceTrait extends  MovieService {
+    def principalsForMovieName(name: String):  List[Principal]
+    def tvSeriesWithGreatestNumberOfEpisodes():  List[TvSeries]
+
+}

src/main/scala/com/mediahub/SpringBootIntegration.scala

@@ -1,13 +0,0 @@
-package com.mediahub
-
-import org.springframework.boot.{CommandLineRunner, SpringApplication}
-import org.springframework.boot.autoconfigure.SpringBootApplication
-
-@SpringBootApplication
-class SpringBootIntegration extends CommandLineRunner{
-  override def run(args: String*): Unit = println("\n\n\n *** Hello World \n\n\n")
-}
-
-object SpringBootIntegration extends App {
-  SpringApplication.run(classOf[SpringBootIntegration])
-}