final

README.md

 # 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.

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

                 principalList = (movieServiceTrait.principalsForMovieName(movieTitle).toList());
             }
 
-            return new ResponseEntity<>(new JsonCustomMapping().mapper(principalList), HttpStatus.OK);
+            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

 
 class JsonCustomMapping extends MovieService {
 
-  def mapper(list:  List[Principal]): String = {
+  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

 import scala.concurrent.Await
 import scala.concurrent.duration.DurationInt
 import scala.language.postfixOps
+import scala.util.Try
 
 
 @Component
       .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
   }
 
-
-  val res = getSeriesTitle(titleEpisodeResource)
-  val r = getTitleByIds(titleBasicsResource, res)
-  println(r)
-
   override def principalsForMovieName(name: String): List[Principal] = {
 
     val titleId: String = getIdOfTitle(titleBasicsResource, name)
     personsList
   }
 
-  override def tvSeriesWithGreatestNumberOfEpisodes(): List[TvSeries] = ???
-    //val personsIdList: List[String] = getSeriestitle(titleEpisodeResource)
+  override def tvSeriesWithGreatestNumberOfEpisodes(): List[TvSeries] = {
+    val seriesTitleMap = getSeriesTitle(titleEpisodeResource)
+    val titleListMaxEpisode = getTitleByIds(titleBasicsResource, seriesTitleMap)
+    titleListMaxEpisode
+  }
 
 
 
     result.value.get.get.get
   }
 
-  def getTitleByIds(file: File, titleIdMap: mutable.Map[String, Int]): Seq[Option[String]] = {
-    val result = fileSource(file)
-      .mapAsync(50) {
-        res =>
-          Source.single(res)
-            .via(lineParser)
-            .filter(row => titleIdMap.contains(row.getOrElse("tconst", "")))
-            .map(a => a.get("primaryTitle"))
-            .withAttributes(supervisionStrategy(resumingDecider))
-            .runWith(Sink.collection)
-      }
-      .withAttributes(supervisionStrategy(resumingDecider))
-      .runWith(Sink.head)
+  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 te = fileSource(file)
+    val result = fileSource(file)
       .mapAsync(50) {
         res =>
           Source.single(res)
       .withAttributes(supervisionStrategy(resumingDecider))
       .runWith(Sink.head)
 
-    Await.result(te, 5 minutes)
+    Await.result(result, 5 minutes)
   }
 
 
       .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"), a("birthYear").toInt, a("deathYear").toIntOption, a.get("primaryProfession").toList))
+      .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)