package com.pinterest.secor.io.impl;

import com.pinterest.secor.common.SecorConfig;
import com.pinterest.secor.io.AdgearReader;
import com.pinterest.secor.io.KeyValue;

import net.minidev.json.JSONObject;
import net.minidev.json.JSONValue;

// Converts delivery JSON to Beh TSV
public class AdgearDeliveryJsonReader implements AdgearReader {
    private final String timestampFieldname;

    public AdgearDeliveryJsonReader(SecorConfig secorConfig) {
        timestampFieldname = secorConfig.getMessageTimestampName();
    }

    public String convert(KeyValue kv) {
        JSONObject jsonObject = null;
        Object value = JSONValue.parse(kv.getValue());
        if (value instanceof JSONObject) {
            jsonObject = (JSONObject) value;
        } else {
            return null;
        }

        Double timestamp = (Double) jsonObject.get(timestampFieldname);
        String cookieId = (String) jsonObject.get("uid");
        Integer segmentId = (Integer) jsonObject.get("segment_id");
        Boolean segmentIsNew = (Boolean) jsonObject.get("segment_new");

        if (timestamp == null || cookieId == null || segmentId == null
            || segmentIsNew == null || !segmentIsNew) {
            return null;
        }

        return String.format("%s\t%d\tseg:%d\n",
                             cookieId, Math.round(timestamp), segmentId);
    }
}

@@ -921,24 +921,88 @@
 get(%22uid%22);%0A
+        Integer buyerId = (Integer) jsonObject.get(%22buyer_id%22);%0A
         Inte
@@ -1162,16 +1162,35 @@
  null %7C%7C
+ buyerId == null %7C%7C
  cookieI
@@ -1350,16 +1350,19 @@
 %25s%5Ct%25d%5Ct
+%25d:
 seg:%25d%5Cn
@@ -1425,16 +1425,25 @@
 estamp),
+ buyerId,
  segment

Write buyer ID when converting delivery logs.