<commit_before>/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package qa.qcri.aidr.collector.collectors;

import java.io.Serializable;
import java.net.SocketException;

//import java.util.logging.Level;
//import java.util.logging.Logger;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.json.JSONException;
import org.json.JSONObject;
import qa.qcri.aidr.collector.beans.AIDR;
import qa.qcri.aidr.collector.beans.CollectionTask;
import qa.qcri.aidr.collector.beans.FetcherResponseToStringChannel;
import qa.qcri.aidr.collector.utils.Config;
import qa.qcri.aidr.collector.logging.Loggable;
import static qa.qcri.aidr.collector.logging.Loggable.LOG_LEVEL;
import qa.qcri.aidr.collector.redis.JedisConnectionPool;
import qa.qcri.aidr.collector.utils.GenericCache;
import qa.qcri.aidr.collector.utils.TwitterStreamQueryBuilder;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.exceptions.JedisConnectionException;
import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterObjectFactory;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;

/**
 *
 * @author Imran
 */
public class TwitterStreamTracker extends Loggable implements Serializable {

    private static Logger logger = Logger.getLogger(TwitterStreamTracker.class.getName());

    private TwitterStream twitterStream;
    private ConfigurationBuilder configBuilder;
    private TwitterStreamQueryBuilder streamQuery;
    private Jedis publisherJedis;
    private String collectionCode; // CrisisID
    private String cacheKey;
    private long counter = 1;
    private String collectionName;

    public TwitterStreamTracker() {
    }

    public TwitterStreamTracker(TwitterStreamQueryBuilder streamFilterQuery, ConfigurationBuilder configurationBuilder, CollectionTask collectionTask) throws Exception {

        this.publisherJedis = JedisConnectionPool.getJedisConnection();
        this.streamQuery = new TwitterStreamQueryBuilder();
        this.collectionCode = collectionTask.getCollectionCode();
        this.collectionName = collectionTask.getCollectionName();
        this.cacheKey = collectionTask.getCollectionCode();
        this.streamQuery = streamFilterQuery;
        this.configBuilder = configurationBuilder;

        GenericCache.getInstance().setTwtConfigMap(cacheKey, collectionTask);
        GenericCache.getInstance().setTwitterTracker(cacheKey, this);
        GenericCache.getInstance().incrCounter(cacheKey, new Long(0));
        collectThroughStreaming();
    }

    private void collectThroughStreaming() {

        StatusListener listener = new StatusListener() {
            JSONObject tweetJSONObject = null;
            CollectionTask collection = GenericCache.getInstance().getTwtConfigMap(getCacheKey());
            JSONObject aidrObject = new JSONObject(new FetcherResponseToStringChannel(new AIDR(getCollectionCode(), getCollectionName(), "twitter")));
            String aidrJson = StringUtils.replace(aidrObject.toString(), "{", ",", 1); // replacing the first occurance of { with ,
            boolean allowAllLanguages = getStreamQuery().isLanguageAllowed(Config.LANGUAGE_ALLOWED_ALL);
            String channelName = Config.FETCHER_CHANNEL + "." + getCollectionCode();
            GenericCache cache = GenericCache.getInstance();

            @Override
            public void onStatus(Status status) {
                String lang = status.getLang();
                if (allowAllLanguages) {
                    publishMessage(status);
                } else {
                    if (getStreamQuery().isLanguageAllowed(lang)) {
                        publishMessage(status);
                    }
                }
            }

            @Override
            public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
            }

            @Override
            public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
                collection.setStatusCode(Config.STATUS_CODE_COLLECTION_RUNNING_WARNING);
                collection.setStatusMessage("Track limitation notice: " + numberOfLimitedStatuses);
                GenericCache.getInstance().setTwtConfigMap(getCacheKey(), collection);
                logger.info(collectionName + ": Track limitation notice: " + numberOfLimitedStatuses);

            }

            @Override
            public void onException(Exception ex) {
                logger.error("Twitter Exception for collection " + collection.getCollectionCode() + " - " + ex.toString());
                //log(LogLevel.WARNING, ex.toString());
                collection.setStatusCode(Config.STATUS_CODE_COLLECTION_ERROR);
            }

            @Override
            public void onScrubGeo(long arg0, long arg1) {
            }

            @Override
            public void onStallWarning(StallWarning arg0) {
                logger.error(collection.getCollectionCode() + " Stall Warning: " + arg0.getMessage());
                //log(LogLevel.WARNING, arg0.toString());
            }

            public void publishMessage(Status status) {
                StringBuilder rawTweetJSON = new StringBuilder(TwitterObjectFactory.getRawJSON(status));
                int length = rawTweetJSON.length();
                if (rawTweetJSON.charAt(length - 1) == '}') {
                    rawTweetJSON.replace(length - 1, length, aidrJson);
                    publisherJedis.publish(channelName, status.getText());
                    counter++;
                    if (counter >= Config.FETCHER_REDIS_COUNTER_UPDATE_THRESHOLD) {<commit_msg>small fix<commit_after>/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package qa.qcri.aidr.collector.collectors;

import java.io.Serializable;
import java.net.SocketException;

//import java.util.logging.Level;
//import java.util.logging.Logger;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.json.JSONException;
import org.json.JSONObject;
import qa.qcri.aidr.collector.beans.AIDR;
import qa.qcri.aidr.collector.beans.CollectionTask;
import qa.qcri.aidr.collector.beans.FetcherResponseToStringChannel;
import qa.qcri.aidr.collector.utils.Config;
import qa.qcri.aidr.collector.logging.Loggable;
import static qa.qcri.aidr.collector.logging.Loggable.LOG_LEVEL;
import qa.qcri.aidr.collector.redis.JedisConnectionPool;
import qa.qcri.aidr.collector.utils.GenericCache;
import qa.qcri.aidr.collector.utils.TwitterStreamQueryBuilder;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.exceptions.JedisConnectionException;
import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterObjectFactory;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;

/**
 *
 * @author Imran
 */
public class TwitterStreamTracker extends Loggable implements Serializable {

    private static Logger logger = Logger.getLogger(TwitterStreamTracker.class.getName());

    private TwitterStream twitterStream;
    private ConfigurationBuilder configBuilder;
    private TwitterStreamQueryBuilder streamQuery;
    private Jedis publisherJedis;
    private String collectionCode; // CrisisID
    private String cacheKey;
    private long counter = 1;
    private String collectionName;

    public TwitterStreamTracker() {
    }

    public TwitterStreamTracker(TwitterStreamQueryBuilder streamFilterQuery, ConfigurationBuilder configurationBuilder, CollectionTask collectionTask) throws Exception {

        this.publisherJedis = JedisConnectionPool.getJedisConnection();
        this.streamQuery = new TwitterStreamQueryBuilder();
        this.collectionCode = collectionTask.getCollectionCode();
        this.collectionName = collectionTask.getCollectionName();
        this.cacheKey = collectionTask.getCollectionCode();
        this.streamQuery = streamFilterQuery;
        this.configBuilder = configurationBuilder;

        GenericCache.getInstance().setTwtConfigMap(cacheKey, collectionTask);
        GenericCache.getInstance().setTwitterTracker(cacheKey, this);
        GenericCache.getInstance().incrCounter(cacheKey, new Long(0));
        collectThroughStreaming();
    }

    private void collectThroughStreaming() {

        StatusListener listener = new StatusListener() {
            JSONObject tweetJSONObject = null;
            CollectionTask collection = GenericCache.getInstance().getTwtConfigMap(getCacheKey());
            JSONObject aidrObject = new JSONObject(new FetcherResponseToStringChannel(new AIDR(getCollectionCode(), getCollectionName(), "twitter")));
            String aidrJson = StringUtils.replace(aidrObject.toString(), "{", ",", 1); // replacing the first occurance of { with ,
            boolean allowAllLanguages = getStreamQuery().isLanguageAllowed(Config.LANGUAGE_ALLOWED_ALL);
            String channelName = Config.FETCHER_CHANNEL + "." + getCollectionCode();
            GenericCache cache = GenericCache.getInstance();

            @Override
            public void onStatus(Status status) {
                String lang = status.getLang();
                if (allowAllLanguages) {
                    publishMessage(status);
                } else {
                    if (getStreamQuery().isLanguageAllowed(lang)) {
                        publishMessage(status);
                    }
                }
            }

            @Override
            public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
            }

            @Override
            public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
                collection.setStatusCode(Config.STATUS_CODE_COLLECTION_RUNNING_WARNING);
                collection.setStatusMessage("Track limitation notice: " + numberOfLimitedStatuses);
                GenericCache.getInstance().setTwtConfigMap(getCacheKey(), collection);
                logger.info(collectionName + ": Track limitation notice: " + numberOfLimitedStatuses);

            }

            @Override
            public void onException(Exception ex) {
                logger.error("Twitter Exception for collection " + collection.getCollectionCode() + " - " + ex.toString());
                //log(LogLevel.WARNING, ex.toString());
                collection.setStatusCode(Config.STATUS_CODE_COLLECTION_ERROR);
            }

            @Override
            public void onScrubGeo(long arg0, long arg1) {
            }

            @Override
            public void onStallWarning(StallWarning arg0) {
                logger.error(collection.getCollectionCode() + " Stall Warning: " + arg0.getMessage());
                //log(LogLevel.WARNING, arg0.toString());
            }

            public void publishMessage(Status status) {
                StringBuilder rawTweetJSON = new StringBuilder(TwitterObjectFactory.getRawJSON(status));
                int length = rawTweetJSON.length();
                if (rawTweetJSON.charAt(length - 1) == '}') {
                    rawTweetJSON.replace(length - 1, length, aidrJson);
                    publisherJedis.publish(channelName, rawTweetJSON.toString());
                    counter++;
                    if (counter >= Config.FETCHER_REDIS_COUNTER_UPDATE_THRESHOLD) {
{'fraction_non_alphanumeric': 0.0887354337505395, 'fraction_numerical': 0.0034527406128614588, 'mean_word_length': 3.017337031900139, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, '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': '5016517', 'n_tokens_mistral': 3071, 'n_tokens_neox': 2934, 'n_words': 565}