<commit_before>package com.bloomberg.bask.task;

import com.bloomberg.bask.serializer.JSONString;
import com.bloomberg.bask.system.Envelope;
import com.bloomberg.bask.system.SystemProducer;

import java.io.File;
import java.io.IOException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.store.MMapDirectory;
import org.apache.lucene.store.RAMDirectory;
import org.apache.lucene.util.Version;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import uk.co.flax.luwak.CandidateMatcher;
import uk.co.flax.luwak.InputDocument;
import uk.co.flax.luwak.InputDocument.Builder;
import uk.co.flax.luwak.Monitor;
import uk.co.flax.luwak.MonitorQuery;
import uk.co.flax.luwak.Presearcher;
import uk.co.flax.luwak.QueryError;
import uk.co.flax.luwak.QueryMatch;
import uk.co.flax.luwak.presearcher.TermFilteredPresearcher;
import uk.co.flax.luwak.parsers.LuceneQueryCache;
import uk.co.flax.luwak.matchers.SimpleMatcher;

public class SubscriptionTask implements InitableTask, StreamTask {

    private static final Logger logger = LoggerFactory.getLogger(SubscriptionTask.class);

    private Monitor monitor;
    private JSONString decoder;

    @Override
    public void init(Properties config) {
        try {
            decoder = new JSONString();
            String indexPath = config.getProperty("subscription.lucene.directory");
            LuceneQueryCache queryCache = new LuceneQueryCache("text");
            Presearcher presearcher = new TermFilteredPresearcher();
            if (indexPath == null || indexPath.isEmpty()) {
                monitor = new Monitor(queryCache, presearcher, new RAMDirectory());
            } else {
                monitor = new Monitor(queryCache, presearcher, new MMapDirectory(new File(indexPath)));
            }
        } catch(IOException e) {
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void process(Envelope envelope, SystemProducer producer) {
        String stream = envelope.getStream();
        switch (stream) {
            case "alerts":
                handleAlert(decoder.fromString(envelope.getMessage()));
                break;
            case "documents":
                Map<String, Object> document = decoder.fromString(envelope.getMessage());
                if (document.get("id_str") != null && document.get("text") != null) {
                    String matches = decoder.toString(handleMatches(document));
                    producer.send(new Envelope(matches, "matches"));
                }
                break;
            default:
                throw new UnexpectedStreamException("Unexpected stream: " + stream);
        }
    }

    private Map<String, Object> handleMatches(Map<String, Object> message) {
        Map<String, Object> result = new HashMap<String, Object>();
        List<String> keywords = new ArrayList<String>();
        try {
            for (QueryMatch match : matchDocument(message)) {
                keywords.add(match.getQueryId());
            }
            result.put("keywords", keywords);
            result.put("tweet", message);
        } catch (IOException ie) {
            logger.error ("Failed to match", ie);
        }

        return result;
    }

    private void handleAlert(Map<String, ?> message) {
        String alert = (String) message.get("alert");
        String action = (String) message.get("action");
        try {
            switch (action) {
                case "add": addAlert(alert, "text:" + alert); break;
                case "delete": deleteAlert(alert); break;
            }
        } catch (IOException ie) {
            logger.error("Failed to handle alert", ie);
        }
    }

    private void addAlert(String id, String alert) throws IOException {
        MonitorQuery monitorQuery = new MonitorQuery(id, alert);
        try {
            for (QueryError error : monitor.update(monitorQuery)) {
                logger.warn (error.toString());
            }
        } catch (IOException ie) {
            logger.error("Could not add query with id '{}' and value '{}'", id, alert);
            throw ie;
        }
    }

    private void deleteAlert(String id) throws IOException {
        try {
            monitor.deleteById(id);
        } catch (IOException ie) {
            logger.error("Could not delete query with id '{}'", id, ie);
            throw ie;
        }
    }

    private SimpleMatcher matchDocument(Map<String, Object> document) throws IOException {
        InputDocument doc = InputDocument.builder((String) document.get("id_str"))
            .addField ("text", (String) document.get("text"), new StandardAnalyzer(Version.LUCENE_48))
            .build();
        return monitor.match(doc, SimpleMatcher.FACTORY);
    }
}
<commit_msg>Matches now show positions<commit_after>package com.bloomberg.bask.task;

import com.bloomberg.bask.serializer.JSONString;
import com.bloomberg.bask.system.Envelope;
import com.bloomberg.bask.system.SystemProducer;

import java.io.File;
import java.io.IOException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.store.MMapDirectory;
import org.apache.lucene.store.RAMDirectory;
import org.apache.lucene.util.Version;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import uk.co.flax.luwak.CandidateMatcher;
import uk.co.flax.luwak.InputDocument;
import uk.co.flax.luwak.InputDocument.Builder;
import uk.co.flax.luwak.Monitor;
import uk.co.flax.luwak.MonitorQuery;
import uk.co.flax.luwak.Presearcher;
import uk.co.flax.luwak.QueryError;
import uk.co.flax.luwak.QueryMatch;
import uk.co.flax.luwak.intervals.IntervalsMatcher;
import uk.co.flax.luwak.intervals.IntervalsQueryMatch;
import uk.co.flax.luwak.presearcher.TermFilteredPresearcher;
import uk.co.flax.luwak.parsers.LuceneQueryCache;

public class SubscriptionTask implements InitableTask, StreamTask {

    private static final Logger logger = LoggerFactory.getLogger(SubscriptionTask.class);

    private Monitor monitor;
    private JSONString decoder;

    @Override
    public void init(Properties config) {
        try {
            decoder = new JSONString();
            String indexPath = config.getProperty("subscription.lucene.directory");
            LuceneQueryCache queryCache = new LuceneQueryCache("text");
            Presearcher presearcher = new TermFilteredPresearcher();
            if (indexPath == null || indexPath.isEmpty()) {
                monitor = new Monitor(queryCache, presearcher, new RAMDirectory());
            } else {
                monitor = new Monitor(queryCache, presearcher, new MMapDirectory(new File(indexPath)));
            }
        } catch(IOException e) {
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void process(Envelope envelope, SystemProducer producer) {
        String stream = envelope.getStream();
        switch (stream) {
            case "alerts":
                handleAlert(decoder.fromString(envelope.getMessage()));
                break;
            case "documents":
                Map<String, Object> document = decoder.fromString(envelope.getMessage());
                if (document.get("id_str") != null && document.get("text") != null) {
                    String matches = decoder.toString(handleMatches(document));
                    producer.send(new Envelope(matches, "matches"));
                }
                break;
            default:
                throw new UnexpectedStreamException("Unexpected stream: " + stream);
        }
    }

    private Map<String, Object> handleMatches(Map<String, Object> message) {
        Map<String, Object> result = new HashMap<String, Object>(), match, position, positions;
        List<Object> matches = new ArrayList<Object>(), fields;
        try {
            for (IntervalsQueryMatch queryMatch : matchDocument(message)) {
                match = new HashMap<String, Object>();
                match.put("queryid", queryMatch.getQueryId());
                positions = new HashMap<String, Object>();
                for (String fieldName : queryMatch.getFields()) {
                    fields = new ArrayList<Object>();
                    for (IntervalsQueryMatch.Hit queryHit : queryMatch.getHits(fieldName)) {
                        position = new HashMap<String, Object>();
                        position.put("endOffset", queryHit.endOffset);
                        position.put("endPosition", queryHit.endPosition);
                        position.put("startOffset", queryHit.startOffset);
                        position.put("startPosition", queryHit.startPosition);
                        fields.add(position);
                    }
                    positions.put(fieldName, fields);
                }
                match.put("positions", positions);
                matches.add(match);
            }
            result.put("matches", matches);
            result.put("tweet", message);
        } catch (IOException ie) {
            logger.error ("Failed to match", ie);
        }

        return result;
    }

    private void handleAlert(Map<String, ?> message) {
        String alert = (String) message.get("alert");
        String action = (String) message.get("action");
        try {
            switch (action) {
                case "add": addAlert(alert, "text:" + alert); break;
                case "delete": deleteAlert(alert); break;
            }
        } catch (IOException ie) {
            logger.error("Failed to handle alert", ie);
        }
    }

    private void addAlert(String id, String alert) throws IOException {
        MonitorQuery monitorQuery = new MonitorQuery(id, alert);
        try {
            for (QueryError error : monitor.update(monitorQuery)) {
                logger.warn (error.toString());
            }
        } catch (IOException ie) {
            logger.error("Could not add query with id '{}' and value '{}'", id, alert);
            throw ie;
        }
    }

    private void deleteAlert(String id) throws IOException {
        try {
            monitor.deleteById(id);
        } catch (IOException ie) {
            logger.error("Could not delete query with id '{}'", id, ie);
            throw ie;
        }
    }

    private IntervalsMatcher matchDocument(Map<String, Object> document) throws IOException {
        InputDocument doc = InputDocument.builder((String) document.get("id_str"))
            .addField ("text", (String) document.get("text"), new StandardAnalyzer(Version.LUCENE_48))
            .build();
        return monitor.match(doc, IntervalsMatcher.FACTORY);
    }
}

{'fraction_non_alphanumeric': 0.11093920640698944, 'fraction_numerical': 0.0007280669821623589, 'mean_word_length': 2.4148539465506524, 'pattern_counts': {'":': 8, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6200483', 'n_tokens_mistral': 3012, 'n_tokens_neox': 2947, 'n_words': 657}