<commit_before>package org.apache.flume.interceptor;

/**
 * Created by koichi.yanagimoto on 2017/05/15.
 */

import java.time.LocalDateTime;
import java.util.*;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.log4j.Logger;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;

public class JsonBodyTimestampInterceptor implements Interceptor {
    private static final Logger LOG;

    static {
        LOG = Logger.getLogger(JsonBodyTimestampInterceptor.class);
    }

    private JsonBodyTimestampInterceptor() {}

    @Override
    public void initialize() {}

    @Override
    public Event intercept(Event event) {

        Map<String, String> headers = event.getHeaders();

        String body = new String(event.getBody());
        JSONParser jsonParser = new JSONParser();
        JSONObject jsonObject = null;

        try {
            jsonObject = (JSONObject) jsonParser.parse(body);
        } catch (org.json.simple.parser.ParseException e) {
            LOG.info("Received this log message that is not formatted in json: "+body+"\n");
            return event;
        }

        String timestamp = jsonObject.get("@timestamp").toString();
        LocalDateTime parsedDateTime = LocalDateTime.parse(timestamp);

        headers.put("timestampYear", String.valueOf(parsedDateTime.getYear()));
        headers.put("timestampMonth", String.valueOf(parsedDateTime.getMonthValue()));
        headers.put("timestampDay", String.valueOf(parsedDateTime.getDayOfMonth()));
        headers.put("timestampHour", String.valueOf(parsedDateTime.getHour()));

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (Event event:events) {
            intercept(event);
        }
        return events;
    }

    @Override
    public void close() {}

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new JsonBodyTimestampInterceptor();
        }

        @Override
        public void configure(Context context) {}
    }
}<commit_msg>move timestamp key from code to property<commit_after>package org.apache.flume.interceptor;

/**
 * Created by koichi.yanagimoto on 2017/05/15.
 */

import java.util.*;
import java.time.LocalDateTime;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.log4j.Logger;

import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;

public class JsonBodyTimestampInterceptor implements Interceptor {
    private static final Logger LOG;
    static {
        LOG = Logger.getLogger(JsonBodyTimestampInterceptor.class);
    }

    private String timestampKey;
    protected Context context;

    public static final String TIMESTAMP_KEY_NAME = "properties.timestampKeyName";

    public JsonBodyTimestampInterceptor(Context context) {
        this.context = context;
        this.timestampKey = context.getString(TIMESTAMP_KEY_NAME);
    }

    @Override
    public void initialize() {}

    @Override
    public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();

        String body = new String(event.getBody());
        JSONParser jsonParser = new JSONParser();
        JSONObject jsonObject = null;

        try {
            jsonObject = (JSONObject) jsonParser.parse(body);
        } catch (org.json.simple.parser.ParseException e) {
            LOG.info("Received this log message that is not formatted in json: "+body+"\n");
            return event;
        }

        String timestamp = jsonObject.get(this.timestampKey).toString();
        LocalDateTime parsedDateTime = LocalDateTime.parse(timestamp);

        headers.put("timestampYear", String.valueOf(parsedDateTime.getYear()));
        headers.put("timestampMonth", String.valueOf(parsedDateTime.getMonthValue()));
        headers.put("timestampDay", String.valueOf(parsedDateTime.getDayOfMonth()));
        headers.put("timestampHour", String.valueOf(parsedDateTime.getHour()));

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (Event event:events) {
            intercept(event);
        }
        return events;
    }

    @Override
    public void close() {}



    public static class Builder implements Interceptor.Builder {
        private Context ctx;
        @Override
        public Interceptor build() {
            return new JsonBodyTimestampInterceptor(ctx);
        }

        @Override
        public void configure(Context context) {
            this.ctx = context;
        }
    }
}
{'fraction_non_alphanumeric': 0.1024537236332329, 'fraction_numerical': 0.0038743004735256135, 'mean_word_length': 2.895222129086337, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '3959192', 'n_tokens_mistral': 1237, 'n_tokens_neox': 1193, 'n_words': 286}