import scala.annotation.tailrec

val data = {
  scala.io.Source.fromFile("war-and-society-1-2-3-4-simple-edges.sorted.csv").getLines()
    .map { l =>
      val arr = l.split(",").map(_.toInt)
      arr(0) -> arr(1)
    }.toArray
}



@tailrec
def processGraph(data: Array[(Int, Int)]): Array[(Int, Int)] = {
  val inputs = data.groupBy(_._2).map { case (k, v) => k -> v.map(_._1) }
  val outputs = data.groupBy(_._1).map { case (k, v) => k -> v.map(_._2) }

  val singleLinks = outputs.filter(_._2.length == 1).filter{case (from,Array(to)) => Array(from).toList == inputs(to).toList }.map { case (from, Array(to)) => from -> to}

  if (singleLinks.isEmpty) {
    return data
  }

  val newData = collection.mutable.HashSet[(Int, Int)]()

  data.foreach { case (from, to) =>
    // нашли связь 1вых->1вх
    if (singleLinks.get(from) == Option(to)) {
      // перенести исходящие узлы удаляемго узла к преды дущему
      outputs.get(to).foreach { outs =>
        outs.foreach { out =>
          newData.add((from, out))
        }
      }
    } else {
      if (!singleLinks.exists { case (f, t) => (t == from)} ) {
        newData.add((from, to))
      }
    }
  }

  processGraph(newData.toArray)

}

processGraph(data)

import scala.annotation.tailrec

/*  
   Делаем замену типа вот так (ребра без стрелок считать направленными слева направо)

           1        6                       1   6                          
            \      /                         \ /                         
        2 -> 4 -> 5 -> 7       ====>     2 -> 4 -> 7                                               
            /      \                         / \                         
          3         8                      3    8      


  Суть реализации - сделать два словаря: для входящих и исходящих ребер каждого из узлов
  Т.е.  4->{1,2,3}, 5->{4}, 6->{5}, 7->{5}, 8->{5} - словать входов для каждого узла
        1->{4}, 2->{4}, 3->{4}, 4->{5}, 5->{6,7,8} - словарь выходов
  
  Потом найдем кандидата на удаление: в первом словаре есть 5->{4}, во втором 4->{5}
  Таким образом ребро 4->5 кандидат на схлопывание 



  Для незнающих синтаксис Scala
  val data : Type - объявление константы (Type data)
  (Int, Int) - кортеж их двух элементов типа  Int (Tuple<Int,Int>)
  _1 - первый элемент кортежа, _2 - второй и т.п. 
  Например: 
    val tuple = ("A","B") 
    tuple._1 ==  "A" // true
    tuple._2 == "B" // true

  Array[Type] - массив элементов типа Type (Type[])
  Map[Int, Array[Int]] - Map<Int,Int[]>
  

*/

// Ориентированный граф представлен в виде множества пар чисел,
// каждая пара - (вершина от которой направлено ребро, вершина к которой направлено ребро)

// вычитаем все ребра из файла в массив типа (Int, Int), преобразуя на ходу в Int
val data: Array[(Int, Int)] = {
  scala.io.Source.fromFile("war-and-society-1-2-3-4-simple-edges.sorted.csv").getLines()
    .map { l =>
      val arr = l.split(",").map(_.toInt)
      arr(0) -> arr(1)
    }.toArray
}

/**
 * Рекурсивная функция удаления пар вершин, связанных единственным ребром
 *
 * @param data - граф в виде массив связей между вершинами
 * @return - граф в виде массив связей между вершинами
 */
@tailrec
def processGraph(data: Array[(Int, Int)]): Array[(Int, Int)] = {

  // Хэш таблица с входящими связями для каждой вершины - вершина -> массив входящих ребер
  // т.е. 4 -> {1,2,3}, 5->{4}, 6->{5}, 7->{5}, 8->{5} 
  val inputs: Map[Int, Array[Int]] = data
    .groupBy(_._2) // группируем по врешине, в которую направлено ребро, на выходе Map[Int, Array[(Int,Int)]
    .map { case (k, v) => k -> v.map(_._1) } // оставляем в занчении  только входящие узлы, т.е. приводим к  Map[Int, Array[Int]]

  // Хэш таблица с исходящими связями для каждой вершины - вершина -> массив исходящих ребер
  // Аналогично inputs, только группировка относительно вершины, из которой выходит ребро
  // т.е. 1->{4}, 2->{4}, 3->{4}, 4->{5}, 5->{6,7,8}
  val outputs: Map[Int, Array[Int]] = data.groupBy(_._1).map { case (k, v) => k -> v.map(_._2) }

  // Пары с одним входом и одним выходом - кандидаты на схлопывание
  // в outputs ищем вершины с единственным выходным ребром, для которых есть пара в inputs тем же ребром, но уже входным
  // т.е. надо найти 5->{4} из inputs и 4->{5} из outputs
  val singleLinks: Map[Int, Int] =
    outputs
      .filter(_._2.length == 1) // ищем вершины с одним выходом (т.е. такую верщину, у ктр. длина массива ввыходов == 1)
      .filter { case (from, Array(to)) => // ищем к найденой вершиние с одним выходом пару из хэш таблицы входов, такую, чтобы выход одной вершины == вход другой 
        Array(from).toList == inputs(to).toList 
      } 
      .map { case (from, Array(to)) => from -> to } // преобразуем из Map[Int, Array[Int]] в Map[Int, Int], т.к. связаннре ребро одно и массив не нужен
  
  // singleLinks = 4->5

  // если пар для удаления нет, от делать больше нечего, возвращаем исходный граф
  if (singleLinks.isEmpty) {
    return data
  }

  // сюда будем собирать новый схлопнутый граф, Set - для перестраховки от дубляжей
  val newData = collection.mutable.HashSet[(Int, Int)]()
  
  /* 
     Теперь перебираем все ребра графа и проверяем, можно-ли их удалить,
     т.е. входит-ли данное ребро в singleLinks
     Если ребро является кандидатом на удаление, то сохраняем только вершину из которой направлено ребро
     и перевпривязываем к ней выходые ребра второй вершины
     Если ребра в singleLinks нет,  то просто переносим его в результирующий граф 
                        
  */  
  data.foreach { case (from, to) =>
    // нашли пару вершин - кандидатов на удаление (4->5)
    if (singleLinks.get(from) == Option(to)) {
      // перепривязываем выходые ребра второй вершины к первой
      // т.е. вновь образованные пары 4->6, 4->7, 4->8
      outputs.get(to).foreach { outs =>
        outs.foreach { out =>
          newData.add((from, out))
        }
      }
    } else { // иначе просто переносим ребро в результат
      // проверка, чтобы исх ребра из удаленной вершины 
      // (которые мы переприцепили к вервой) не попали в результат
      // т.е. пары 5->6, 5->7, 5->8 нам уже не нужны
      if (!singleLinks.exists { case (f, t) => (t == from) }) { 
        newData.add((from, to))
      }
    }
  }

  processGraph(newData.toArray)

}

processGraph(data)