working finder service for people depending on title name

src/main/resources/test.tsv

@@ -1,17 +0,0 @@
-tconst	titleType	primaryTitle	originalTitle	isAdult	startYear	endYear	runtimeMinutes	genres
-tt0000001	short	Carmencita	Carmencita	0	1894	\N	1	Documentary,Short
-tt0000002	short	Le clown et ses chiens	Le clown et ses chiens	0	1892	\N	5	Animation,Short
-tt0000003	short	Pauvre Pierrot	Pauvre Pierrot	0	1892	\N	4	Animation,Comedy,Romance
-tt0000004	short	Un bon bock	Un bon bock	0	1892	\N	12	Animation,Short
-tt0000005	short	Blacksmith Scene	Blacksmith Scene	0	1893	\N	1	Comedy,Short
-tt0000006	short	Chinese Opium Den	Chinese Opium Den	0	1894	\N	1	Short
-tt0000007	short	Corbett and Courtney Before the Kinetograph	Corbett and Courtney Before the Kinetograph	0	1894	\N	1	Short,Sport
-tt0000008	short	Edison Kinetoscopic Record of a Sneeze	Edison Kinetoscopic Record of a Sneeze	0	1894	\N	1	Documentary,Short
-tt0000009	movie	Miss Jerry	Miss Jerry	0	1894	\N	45	Romance
-tt0000010	short	Leaving the Factory	La sortie de l'usine Lumi?¿re ?á Lyon	0	1895	\N	1	Documentary,Short
-tt0000011	short	Akrobatisches Potpourri	Akrobatisches Potpourri	0	1895	\N	1	Documentary,Short
-tt0000012	short	The Arrival of a Train	L'arriv?®e d'un train ?á La Ciotat	0	1896	\N	1	Documentary,Short
-tt0000013	short	The Photographical Congress Arrives in Lyon	Le d?®barquement du congr?¿s de photographie ?á Lyon	0	1895	\N	1	Documentary,Short
-tt0000014	short	The Waterer Watered	L'arroseur arros?®	0	1895	\N	1	Comedy,Short
-tt0000015	short	Autour d'une cabine	Autour d'une cabine	0	1894	\N	2	Animation,Short
-tt0000016	short	Boat Leaving the Port	Barque sortant du port	0	1895	\N	1	Documentary,Short

src/main/resources/test2.tsv

@@ -1,2 +0,0 @@
-tconst	titleType	name	originalTitle	isAdult	startYear	endYear	runtimeMinutes	ge
-tt0000015	short	Antoine	Autour d'une cabine	0	1900	\N	2	Animation,Short

src/main/scala/com/mediahub/FileReader.scala

@@ -1,24 +1,19 @@
 package com.mediahub
 
-import akka.{Done, NotUsed}
-import org.springframework.stereotype.Component
-
-import java.io.File
+import akka.NotUsed
 import akka.actor.ActorSystem
 import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap}
 import akka.stream.javadsl.FileIO
-import akka.stream.scaladsl.{Compression, Flow, Framing, Keep, Sink, Source}
+import akka.stream.scaladsl.{Compression, Framing, Sink, Source}
 import akka.util.ByteString
+import org.springframework.stereotype.Component
 
+import java.io.File
 import java.nio.file.Paths
-import java.time.Duration
-import java.util.stream.Collectors
 import scala.collection.immutable
-import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.DurationInt
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.language.postfixOps
-import scala.util.{Failure, Success}
 
 
 @Component
@@ -26,73 +21,60 @@ class FileReader {
 
   implicit val system: ActorSystem = ActorSystem()
 
-  val list: List[File] = getListOfFiles
-  val source: Source[File, NotUsed] = Source(list)
+  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 titleId: String = getIdOfTitle(titleBasicsResource, "Carmencita")
+  println("value ", titleId)
 
-  val result: Future[immutable.Iterable[Option[String]]] = source
-    .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
-    .via(CsvParsing.lineScanner(CsvParsing.Tab))
-    .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
-    .filter(row => row.getOrElse("primaryTitle", "") == "Autour d'une cabine")
-    .map(a => a.get("tconst"))
-    .runWith(Sink.collection)
+  val personsIdList: List[String]  = getIdOfPersons(titlePrincipalsResource, titleId)
+  println("value ", personsIdList)
 
-  result.onComplete {
-    case Failure(exception) => println(s"failure: $exception")
+  val personsList: List[String] = getPersons(nameBasicsResource, personsIdList)
+  println("value ", personsList)
 
-    case Success(value) => {
-      println(s"sucess: $value")
-    }
 
+  def getIdOfTitle(file: File, titleName: String): String = {
+    val result: Future[Option[String]] = 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(headerPlaceholder = Option.empty))
+      .filter(row => row.getOrElse("primaryTitle", "") == titleName)
+      .map(a => a.get("tconst"))
+      .runWith(Sink.head)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.get
   }
-  Await.result(result,  5 hours)
-
-  val re: List[Option[String]] = result.value.get.get.toList
-
-  println("value ", re.iterator.next().get)
-
-  //.runWith(Sink.seq)
-/*
-  source
-    .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
-    .via(CsvParsing.lineScanner(CsvParsing.Tab))
-    .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
-    .runWith(Sink.collection)
-    .foreach(a => println("result", a))
-*/
-
-  /* .filter(row => row.getOrElse("tconst", "name") == (result.value, "Antoine"))
-   .runForeach(println)*/
-
-
-  // readAsync()
-
-
-  /* def readAsync(): Source = {
-     Source(list)
-       .mapAsync(4) { value =>
-         implicit val ec: ExecutionContext = ActorSystem().dispatcher
-         Future {
-           source
-             .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
-             .via(CsvParsing.lineScanner(CsvParsing.Tab))
-             .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
-             .filter(row => row.getOrElse("primaryTitle", "") == "Autour d'une cabine")
-             .runForeach(println)
-           println(s"Process in Thread:${Thread.currentThread().getName}")
-           value
-         }
-       }
-       .runWith(Sink.foreach(value => println(s"Item emitted:$value in Thread:${Thread.currentThread().getName}")))
-   }*/
-
-  def getListOfFiles: List[File] = {
-    val d = new File("src/main/resources")
-    if (d.exists && d.isDirectory) {
-      d.listFiles.filter(_.isFile).toList
-    } else {
-      List[File]()
-    }
+
+
+  def getIdOfPersons(file: File, titleId: String): List[String] = {
+    val result: Future[immutable.Iterable[Option[String]]] = 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(headerPlaceholder = Option.empty))
+      .filter(row => row.getOrElse("tconst", "") == titleId)
+      .map(a => a.get("nconst"))
+      .runWith(Sink.collection)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.toList.flatten
   }
 
+  def getPersons(file: File, personsIdList: List[String]): List[String] = {
+    val result: Future[immutable.Iterable[Option[String]]] = 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(headerPlaceholder = Option.empty))
+      .filter(row => personsIdList.contains(row.getOrElse("nconst", "")))
+      .map(a => a.get("primaryName"))
+      .runWith(Sink.collection)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.toList.flatten
+  }
 }