package jobs.textanalyzer;

import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
import java.util.zip.CRC32;

import javax.persistence.EntityManager;

import models.Dream;

import org.apache.commons.io.IOUtils;
import org.tartarus.snowball.SnowballStemmer;

import play.Play;
import play.db.jpa.JPA;
import play.jobs.Every;

@Every("10s")
public class TextComparatorJob extends TextAnalyzerJob {
    private static final int MIN_PERCENT_FOR_SIMILARITY = Integer.parseInt(Play.configuration
            .getProperty("analyzing.min-percent-to-similar"));

    @Override
    public void doJobInternal() {
        List<Dream> dreams = Dream.findAllPublicWithLanguage();
        dreams.parallelStream().forEach(
                dream -> {
                    dream.similarDreams = null;
                    dreams.parallelStream()
                            .filter(otherDream -> !(otherDream.id.equals(dream.id))
                                    || otherDream.user.id.equals(dream.user.id)).collect(Collectors.toList())
                            .parallelStream().forEach(dreamToCompare -> {
                                if (dreamsAreSimilar(dream, dreamToCompare)) {
                                    updateSimilarDreams(dream, dreamToCompare);
                                }
                            });
                });
    }

    private boolean dreamsAreSimilar(Dream dream, Dream dreamToCompare) {
        if (dream.guessedLanguage != dreamToCompare.guessedLanguage) {
            return false;
        }
        Set<String> dreamWords = getStemmedWords(dream.guessedLanguage, dream.dream);
        Set<String> dreamToCompareWords = getStemmedWords(dreamToCompare.guessedLanguage, dreamToCompare.dream);

        Set<Long> dreamShingles = shingles(dreamWords);
        Set<Long> dreamToCompareShingles = shingles(dreamToCompareWords);

        Set<Long> intersect = new TreeSet<Long>(dreamShingles);
        intersect.retainAll(dreamToCompareShingles);
        int same = intersect.size();
        int percent = (int) ((same * 2 / (float) (dreamShingles.size() + dreamToCompareShingles.size())) * 100);
        return percent >= MIN_PERCENT_FOR_SIMILARITY;
    }

    private Set<Long> shingles(Set<String> dreamWords) {
        CRC32 crc32 = new CRC32();
        SortedSet<Long> shingles = new TreeSet<Long>();
        dreamWords.forEach(word -> {
            crc32.update(word.getBytes());
            long crc32Value = crc32.getValue();
            shingles.add(crc32Value);
            crc32.reset();
        });
        return shingles;
    }

    private Set<String> getStemmedWords(String lang, String text) {
        Set<String> words = new HashSet<>();
        SnowballStemmer stemmer = StemmerFactory.getStemmerByLang(Lang.valueOf(lang));
        List<String> stopWords = loadStopWordsDictionary(lang);
        Matcher matcher = WORD_PATTERN.matcher(text);
        while (matcher.find()) {
            String word = matcher.group().toLowerCase();
            if (!stopWords.contains(word)) {
                stemmer.setCurrent(word);
                stemmer.stem();
                words.add(stemmer.getCurrent());
            }
        }
        return words;
    }

    private List<String> loadStopWordsDictionary(String guessedLanguage) {
        InputStream is = Play.classloader.getResourceAsStream("jobs/textanalyzer/dictionary/stop/stop_"
                + guessedLanguage + ".txt");
        try {
            List<String> dictionary = IOUtils.readLines(is);
            return dictionary;
        } catch (IOException e) {
        } finally {
            IOUtils.closeQuietly(is);
        }
        return null;
    }

    private void updateSimilarDreams(Dream dream, Dream dreamToCompare) {
        Map<Long, Set<Long>> usersWithSimilarDreams = new HashMap<>();
        String similarDreams = dream.similarDreams;
        if (similarDreams != null) {
            usersWithSimilarDreams = dream.convertSimilarDreams();
        }
        Set<Long> dreamsIDS = usersWithSimilarDreams.get(dreamToCompare.user.id);
        if (dreamsIDS == null) {
            dreamsIDS = new HashSet<>();
        }
        dreamsIDS.add(dreamToCompare.id);
        usersWithSimilarDreams.put(dreamToCompare.user.id, dreamsIDS);
        dream.convertSimilarDreams(usersWithSimilarDreams);

        saveDream(dream);
    }

    private void saveDream(Dream dream) {
        EntityManager em = JPA.newEntityManager();
        JPA.bindForCurrentThread(JPA.DEFAULT, em, false);
        JPA.startTx(JPA.DEFAULT, false);
        Dream dreamToUpdate = Dream.findById(dream.id);
        dreamToUpdate.similarDreams = dream.similarDreams;
        dreamToUpdate.save();
        JPA.closeTx(JPA.DEFAULT);
    }
}

@@ -303,94 +303,17 @@
 til.
-stream.Collectors;%0Aimport java.util.zip.CRC32;%0A%0Aimport javax.persistence.EntityManager
+zip.CRC32
 ;%0A%0Ai
@@ -440,32 +440,8 @@
 ay;%0A
-import play.db.jpa.JPA;%0A
 impo
@@ -813,33 +813,25 @@
      dreams.
-parallelS
+s
 tream().forE
@@ -937,25 +937,17 @@
  dreams.
-parallelS
+s
 tream()%0A
@@ -1030,16 +1030,16 @@
 am.id))%0A
+
         
@@ -1115,83 +1115,8 @@
 d)).
-collect(Collectors.toList())%0A                            .parallelStream().
 forE
@@ -3854,174 +3854,8 @@
 ();%0A
-        String similarDreams = dream.similarDreams;%0A        if (similarDreams != null) %7B%0A            usersWithSimilarDreams = dream.convertSimilarDreams();%0A        %7D%0A
     
@@ -3928,24 +3928,24 @@
 e.user.id);%0A
+
         if (
@@ -4202,401 +4202,19 @@
     
-saveDream(dream);%0A    %7D%0A%0A    private void saveDream(Dream dream) %7B%0A        EntityManager em = JPA.newEntityManager();%0A        JPA.bindForCurrentThread(JPA.DEFAULT, em, false);%0A        JPA.startTx(JPA.DEFAULT, false);%0A        Dream dreamToUpdate = Dream.findById(dream.id);%0A        dreamToUpdate.similarDreams = dream.similarDreams;%0A        dreamToUpdate.save();%0A        JPA.closeTx(JPA.DEFAULT
+dream.save(
 );%0A 

Fix text comparator job connections