// Core Part about Movie Recommendations
// at Danube.co.uk
//===========================================

object CW7b {

import io.Source
import scala.util._


// (1) Implement the function get_csv_url which takes an url-string
//     as argument and requests the corresponding file. The two urls
//     of interest are ratings_url and movies_url, which correspond 
//     to CSV-files.
//
//     The function should ReTurn the CSV-file appropriately broken
//     up into lines, and the first line should be dropped (that is without
//     the header of the CSV-file). The result is a list of strings (lines
//     in the file).

def get_csv_url(url: String) : List[String] = {
  val site = Source.fromURL(url, "ISO-8859-1")
  val site_string = site.mkString
  val output = (site_string.split("\n")).toList
  output.tail
}

  // get_csv_url("https://nms.kcl.ac.uk/christian.urban/ratings.csv")

//val ratings_url = """https://nms.kcl.ac.uk/christian.urban/ratings.csv"""
//val movies_url = """https://nms.kcl.ac.uk/christian.urban/movies.csv"""

// testcases
//-----------
//:
//val movies = get_csv_url(movies_url)
  // val ratings = get_csv_url(ratings_url)

//ratings.length  // 87313
//movies.length   // 9742


// (2) Implement two functions that process the CSV-files from (1). The ratings
//     function filters out all ratings below 4 and ReTurns a list of 
//     (userID, movieID) pairs. The movies function just ReTurns a list 
//     of (movieID, title) pairs. Note the input to these functions, that is
//     the argument lines, will be the output of the function get_csv_url.


def process_ratings(lines: List[String]) : List[(String, String)] = {
  val filter = lines.filter(_.last.asDigit >=4)
  val output = (for(i <- 0 until filter.length) yield ((filter(i).split(",").toList)(0), (filter(i).split(",").toList)(1))).toList
  output
}

def process_movies(lines: List[String]) : List[(String, String)] = {
  val output = (for(i <- 0 until lines.length) yield ((lines(i).split(",").toList)(0), (lines(i).split(",").toList)(1))).toList
  output
}



def process_ratings2(lines: List[String]) : List[(String, String)] = {
  for (cols <- lines.map(_.split(",").toList); 
       if (cols(2).toFloat >= 4)) yield (cols(0), cols(1))  
}

def process_movies2(lines: List[String]) : List[(String, String)] = {
  for (cols <- lines.map(_.split(",").toList)) yield (cols(0), cols(1))  
}

// testcases
//-----------
//val good_ratings = process_ratings(ratings)
//val movie_names = process_movies(movies)

//good_ratings.length   //48580
//movie_names.length    // 9742




// (3) Implement a grouping function that calculates a Map
//     containing the userIDs and all the corresponding recommendations 
//     (list of movieIDs). This  should be implemented in a tail
//     recursive fashion, using a Map m as accumulator. This Map m
//     is set to Map() at the beginning of the calculation.

val ratings_url = """https://nms.kcl.ac.uk/christian.urban/ratings.csv"""
val ratings = get_csv_url(ratings_url)
val good_ratings = process_ratings(ratings)
val v515 = good_ratings.filter(_._1 == "515")
val v515_2 = v515.map(_._2)

def groupById(ratings: List[(String, String)], 
              m: Map[String, List[String]]) : Map[String, List[String]] = {
val users = (for((k,v) <- ratings) yield k).distinct
val movie_ids = (for(i <- 1 to users.length) yield
  (for ((k,v) <- ratings if(i.toString == k)) yield v).toList).toList
  val out_map = (users zip movie_ids).toMap
out_map
}

def groupById2(ratings: List[(String, String)], 
              m: Map[String, List[String]]) : Map[String, List[String]] = ratings match {
  case Nil => m
  case (id, mov) :: rest => {
    val old_ratings = m.getOrElse (id, Nil)
    val new_ratings = m + (id -> (mov :: old_ratings))
    groupById2(rest, new_ratings)
  }
}

val ls0_urban = 
  List(("1", "a"), ("1", "c"), ("1", "c"))

groupById(ls0_urban, Map())
groupById2(ls0_urban, Map())

val ls00_urban = 
  List(("3", "a"), ("3", "c"), ("3", "c"))

groupById(ls00_urban, Map())
groupById2(ls00_urban, Map())

groupById(good_ratings, Map()).getOrElse("515", Nil)
groupById2(good_ratings, Map()).getOrElse("515", Nil)

val ls1_urban = 
  List(("1", "a"), ("2", "a"), 
       ("1", "c"), ("2", "a"), ("1", "c"))

groupById(ls1_urban, Map())
groupById2(ls1_urban, Map())

val ls2_urban = 
  List(("1", "a"), ("1", "b"), ("2", "x"), 
       ("3", "a"), ("2", "y"), ("3", "c"))

groupById(ls2_urban, Map())
groupById2(ls2_urban, Map())

val ls3_urban = (1 to 1000 by 10).map(_.toString).toList
val ls4_urban = ls3_urban zip ls3_urban.tail
val ls5_urban = ls4_urban ::: ls4_urban.reverse

groupById(ls5_urban, Map()) == groupById2(ls5_urban, Map())

groupById(ls5_urban, Map())
groupById2(ls5_urban, Map())

groupById(v515, Map())
groupById2(v515, Map())

groupById(v515.take(1), Map())
groupById2(v515.take(2), Map())

// testcases
//-----------
//val ratings_map = groupById(good_ratings, Map())
//val movies_map = movie_names.toMap

//ratings_map.get("414").get.map(movies_map.get(_)).length
//    => most prolific recommender with 1227 positive ratings

//ratings_map.get("475").get.map(movies_map.get(_)).length
//    => second-most prolific recommender with 787 positive ratings

//ratings_map.get("214").get.map(movies_map.get(_)).length 
//    => least prolific recommender with only 1 positive rating


// (4) Implement a function that takes a ratings map and a movie_name as argument.
//     The function calculates all suggestions containing
//     the movie in its recommendations. It ReTurns a list of all these
//     recommendations (each of them is a list and needs to have the movie deleted, 
//     otherwise it might happen we recommend the same movie).


def favourites(m: Map[String, List[String]], mov: String) : List[List[String]] = {
 (for((k,v) <- m if (v.contains(mov))) yield v.filter(_!=mov).toList).toList
}

def favourites2(m: Map[String, List[String]], mov: String) : List[List[String]] = 
  (for (id <- m.keys.toList;
        if m(id).contains(mov)) yield m(id).filter(_ != mov))


// testcases
//-----------
// movie ID "912" -> Casablanca (1942)
//          "858" -> Godfather
//          "260" -> Star Wars: Episode IV - A New Hope (1977)

//favourites(ratings_map, "912").length  // => 80

// That means there are 80 users that recommend the movie with ID 912.
// Of these 80  users, 55 gave a good rating to movie 858 and
// 52 a good rating to movies 260, 318, 593.



// (5) Implement a suggestions function which takes a rating
//     map and a movie_name as arguments. It calculates all the recommended
//     movies sorted according to the most frequently suggested movie(s) first.

def suggestions(recs: Map[String, List[String]], 
                mov_name: String) : List[String] = {
  val flat = favourites(recs, mov_name).flatten.groupMapReduce(identity)(_ => 1)(_ + _)
  val sorted = flat.toList.sortWith(_._2 > _._2).map(_._1)
  sorted
}


def mapValues[S, T, R](m: Map[S, T], f: T => R) =
  m.map { case (x, y) => (x, f(y)) }

def suggestions2(recs: Map[String, List[String]], 
                    mov_name: String) : List[String] = {
  val favs = favourites(recs, mov_name).flatten
  val favs_counted = mapValues(favs.groupBy(identity), (v:List[String]) => v.size).toList
  val favs_sorted = favs_counted.sortBy(_._2).reverse
  favs_sorted.map(_._1)
}

// testcases
//-----------

//suggestions(ratings_map, "912")
//suggestions(ratings_map, "912").length  
// => 4110 suggestions with List(858, 260, 318, 593, ...)
//    being the most frequently suggested movies



// (6) Implement a recommendations function which generates at most
//     *two* of the most frequently suggested movies. It ReTurns the 
//     actual movie names, not the movieIDs.


def recommendations(recs: Map[String, List[String]],
                    movs: Map[String, String],
                    mov_name: String) : List[String] = {
  val sugg = suggestions(recs, mov_name)
  val movies = (for (i <- 0 until 2 if (i < sugg.length)) yield movs(sugg(i))).toList
  movies
}



// testcases
//-----------
// recommendations(ratings_map, movies_map, "912")
//   => List(Godfather, Star Wars: Episode IV - A NewHope (1977))

//recommendations(ratings_map, movies_map, "260")
//   => List(Star Wars: Episode V - The Empire Strikes Back (1980), 
//           Star Wars: Episode VI - Return of the Jedi (1983))

// recommendations(ratings_map, movies_map, "2")
//   => List(Lion King, Jurassic Park (1993))

// recommendations(ratings_map, movies_map, "0")
//   => Nil

// recommendations(ratings_map, movies_map, "1")
//   => List(Shawshank Redemption, Forrest Gump (1994))

// recommendations(ratings_map, movies_map, "4")
//   => Nil  (there are three ratings for this movie in ratings.csv but they are not positive)     



// (7) Calculate the recommendations for all movies according to
// what the recommendations function in (6) produces (this
// can take a few seconds). Put all recommendations into a list 
// (of strings) and count how often the strings occur in
// this list. This produces a list of string-int pairs,
// where the first component is the movie name and the second
// is the number of how many times the movie was recommended. 
// Sort all the pairs according to the number
// of times they were recommended (most recommended movie name 
// first).

def most_recommended(recs: Map[String, List[String]],
                     movs: Map[String, String]) : List[(String, Int)] = {
  val movies = (((for((k,v) <- movs) yield recommendations(recs, movs, k)).toList).flatten).groupMapReduce(identity)(_ => 1)(_ + _)
  val sorted = movies.toList.sortWith(_._2 > _._2)
  sorted
}

// testcase
//
//most_recommended(ratings_map, movies_map).take(3)
// =>
// List((Matrix,698), 
//      (Star Wars: Episode IV - A New Hope (1977),402), 
//      (Jerry Maguire (1996),382))



}