Adding support for SLA Policies, satisfation_ratings, ticket_metric_events and groups

README.md

@@ -11,7 +11,13 @@ Embulk input plugin for loading [Zendesk](https://www.zendesk.com/) records.
 
 Required Embulk version >= 0.9.6.
 
-**NOTE** This plugin don't support JSON type columns e.g. custom fields, tags, etc for now. But they will be supported soon.
+**NOTE** This plugin differs from original plugin to following:
+* DOES support JSON type columns
+* Supports satisfaction_ratings API
+* Supports ticket_metric_events API
+* Supports SLA Policies API
+* Supports Groups API
+
 
 * **Plugin type**: input
 * **Resume supported**: no

build.gradle

@@ -15,7 +15,7 @@ configurations {
     provided
 }
 
-version = "0.3.6"
+version = "0.4.5"
 
 sourceCompatibility = 1.8
 targetCompatibility = 1.8

gradlew

@@ -168,5 +168,5 @@ eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$A
 if [ "$(uname)" = "Darwin" ] && [ "$HOME" = "$PWD" ]; then
   cd "$(dirname "$0")"
 fi
-
+echo "$JAVACMD" "$@"
 exec "$JAVACMD" "$@"

src/main/java/org/embulk/input/zendesk/ZendeskInputPlugin.java

@@ -23,6 +23,7 @@
 import org.embulk.input.zendesk.services.ZendeskService;
 import org.embulk.input.zendesk.services.ZendeskSupportAPIService;
 import org.embulk.input.zendesk.services.ZendeskUserEventService;
+import org.embulk.input.zendesk.services.ZendeskNormalServices;
 import org.embulk.input.zendesk.utils.ZendeskConstants;
 import org.embulk.input.zendesk.utils.ZendeskDateUtils;
 import org.embulk.input.zendesk.utils.ZendeskUtils;
@@ -173,7 +174,8 @@ public ConfigDiff transaction(final ConfigSource config, final Control control)
 
         // For non-incremental target, we will split records based on number of pages. 100 records per page
         // In preview, run with taskCount = 1
-        if (!Exec.isPreview() && !getZendeskService(task).isSupportIncremental() && getZendeskService(task) instanceof ZendeskSupportAPIService) {
+        if (!Exec.isPreview() && !getZendeskService(task).isSupportIncremental() && getZendeskService(task) instanceof ZendeskSupportAPIService && !Target.SATISFACTION_RATINGS.equals(task.getTarget())) {
+
             final JsonNode result = getZendeskService(task).getDataFromPath("", 0, false, 0);
             if (result.has(ZendeskConstants.Field.COUNT) && !result.get(ZendeskConstants.Field.COUNT).isNull()
                     && result.get(ZendeskConstants.Field.COUNT).isInt()) {
@@ -360,6 +362,9 @@ protected ZendeskService dispatchPerTarget(ZendeskInputPlugin.PluginTask task)
             case TICKET_EVENTS:
             case TICKET_FORMS:
             case TICKET_FIELDS:
+            case TICKET_METRIC_EVENTS:
+            case SATISFACTION_RATINGS:
+            case SLA_POLICIES:
                 return new ZendeskSupportAPIService(task);
             case RECIPIENTS:
             case SCORES:
@@ -369,6 +374,8 @@ protected ZendeskService dispatchPerTarget(ZendeskInputPlugin.PluginTask task)
                 return new ZendeskCustomObjectService(task);
             case USER_EVENTS:
                 return new ZendeskUserEventService(task);
+            case GROUPS:
+                return new ZendeskSupportAPIService(task);
             default:
                 throw new ConfigException("Unsupported " + task.getTarget() + ", supported values: '" + Arrays.toString(Target.values()) + "'");
         }

src/main/java/org/embulk/input/zendesk/clients/ZendeskRestClient.java

@@ -235,6 +235,8 @@ private HttpRequestBase createGetRequest(final String url, final PluginTask task
         final HttpGet request = new HttpGet(url);
         final ImmutableMap<String, String> headers = buildAuthHeader(task);
         headers.forEach(request::setHeader);
+
+//        headers.values().forEach(logger::info);
         return request;
     }
 

src/main/java/org/embulk/input/zendesk/models/Target.java

@@ -12,7 +12,9 @@ public enum Target
     */
     TICKETS("tickets"), USERS("users"), ORGANIZATIONS("organizations"), TICKET_EVENTS("ticket_events"),
     TICKET_METRICS("metric_sets"), TICKET_FIELDS("ticket_fields"), TICKET_FORMS("ticket_forms"),
-    RECIPIENTS("recipients"), SCORES("responses"), OBJECT_RECORDS("data"), RELATIONSHIP_RECORDS("data"), USER_EVENTS("data");
+    TICKET_METRIC_EVENTS("ticket_metric_events"), SATISFACTION_RATINGS("satisfaction_ratings"),
+    SLA_POLICIES("sla_policies"), SCORES("responses"), OBJECT_RECORDS("data"),
+    RECIPIENTS("recipients"), RELATIONSHIP_RECORDS("data"), USER_EVENTS("data"), GROUPS("groups");
 
     String jsonName;
 

src/main/java/org/embulk/input/zendesk/services/ZendeskNPSService.java

@@ -15,6 +15,11 @@ public boolean isSupportIncremental()
     {
         return true;
     }
+    @Override
+    protected String buildURI(final int page, final long startTime, final long endTime)
+    {
+        return buildURI(page, startTime);
+    }
 
     @Override
     protected String buildURI(final int page, final long startTime)

src/main/java/org/embulk/input/zendesk/services/ZendeskNormalServices.java

@@ -49,23 +49,34 @@ public TaskReport addRecordToImporter(final int taskIndex, final RecordImporter
             importDataForIncremental(task, recordImporter, taskReport);
         }
         else {
-            importDataForNonIncremental(task, taskIndex, recordImporter);
+            if(task.getTarget().equals(Target.SATISFACTION_RATINGS)){
+                importDataForNonIncremental(task, taskIndex, recordImporter, true);
+            }
+            else{
+                importDataForNonIncremental(task, taskIndex, recordImporter, false);
+            }
         }
 
         return taskReport;
     }
-
     public JsonNode getDataFromPath(String path, final int page, final boolean isPreview, final long startTime)
+    {
+        return getDataFromPath(path, page,isPreview, startTime, 0);
+    }
+
+    public JsonNode getDataFromPath(String path, final int page, final boolean isPreview, final long startTime, final long endTime)
     {
         if (path.isEmpty()) {
-            path = buildURI(page, startTime);
+            path = buildURI(page, startTime, endTime);
         }
 
         final String response = getZendeskRestClient().doGet(path, task, isPreview);
         return ZendeskUtils.parseJsonObject(response);
     }
 
     protected abstract String buildURI(int page, long startTime);
+    protected abstract String buildURI(int page, long startTime, long endTime);
+
 
     @VisibleForTesting
     protected ZendeskRestClient getZendeskRestClient()
@@ -101,26 +112,33 @@ private void importDataForIncremental(final ZendeskInputPlugin.PluginTask task,
             );
 
             long apiEndTime = 0;
+            int afterStartTimeIndex = 11;
             while (true) {
                 int recordCount = 0;
 
                 // Page argument isn't used in incremental API so we just set it to 0
                 final JsonNode result = getDataFromPath("", 0, false, startTime);
                 final Iterator<JsonNode> iterator = ZendeskUtils.getListRecords(result, task.getTarget().getJsonName());
-                apiEndTime = result.get(ZendeskConstants.Field.END_TIME).asLong();
 
+                if (result.has(ZendeskConstants.Field.END_TIME)){
+                    apiEndTime = result.get(ZendeskConstants.Field.END_TIME).asLong();
+                }
+                else{
+                    if(result.has(ZendeskConstants.Field.NEXT_PAGE)){
+                        String next_page = result.get(ZendeskConstants.Field.NEXT_PAGE).textValue();
+                        apiEndTime = Long.parseLong(next_page.substring(next_page.indexOf(ZendeskConstants.Field.START_TIME)+afterStartTimeIndex));
+                    }
+                }
+                logger.info("Api END Time = '{}'", apiEndTime);
                 int numberOfRecords = 0;
                 if (result.has(ZendeskConstants.Field.COUNT)) {
                     numberOfRecords = result.get(ZendeskConstants.Field.COUNT).asInt();
+                    logger.info("Number of Records {}", numberOfRecords);
                 }
 
                 while (iterator.hasNext()) {
                     final JsonNode recordJsonNode = iterator.next();
 
-                    if (isUpdatedBySystem(recordJsonNode, startTime)) {
-                        continue;
-                    }
-
                     // Contain some records  that later than end_time. Checked and don't add.
                     // Because the api already sorted by updated_at or timestamp for ticket_events, we just need to break no need to check further.
                     if (apiEndTime > endTime) {
@@ -140,6 +158,7 @@ private void importDataForIncremental(final ZendeskInputPlugin.PluginTask task,
                         }
 
                         if (checkedTime > endTime) {
+                            logger.info("Lets break since checked time > endTime {}", endTime);
                             break;
                         }
                     }
@@ -169,6 +188,8 @@ private void importDataForIncremental(final ZendeskInputPlugin.PluginTask task,
                         : apiEndTime;
 
                 if (numberOfRecords < ZendeskConstants.Misc.MAXIMUM_RECORDS_INCREMENTAL || startTime > endTime) {
+                    logger.info("Break because number of Records < MAXIMUM RECORDS {}", numberOfRecords);
+                    logger.info("or start time {}> end Time {}", startTime, endTime);
                     break;
                 }
             }
@@ -239,13 +260,13 @@ private void fetchSubResourceAndAddToImporter(final JsonNode jsonNode, final Zen
                 }
             }
             catch (final ConfigException e) {
+                logger.info("Some exception for id {}", jsonNode.get("id"));
                 // Sometimes we get 404 when having invalid endpoint, so ignore when we get 404 InvalidEndpoint
                 if (!(e.getCause() instanceof ZendeskException && ((ZendeskException) e.getCause()).getStatusCode() == HttpStatus.SC_NOT_FOUND)) {
                     throw e;
                 }
             }
         });
-
         recordImporter.addRecord(jsonNode);
     }
 
@@ -269,18 +290,68 @@ private boolean isUpdatedBySystem(final JsonNode recordJsonNode, final long star
         return false;
     }
 
-    private void importDataForNonIncremental(final ZendeskInputPlugin.PluginTask task, final int taskIndex, RecordImporter recordImporter)
+    private void importDataForNonIncremental(final ZendeskInputPlugin.PluginTask task,  final int taskIndex, RecordImporter recordImporter, final boolean getStartDate)
     {
-        // Page start from 1 => page = taskIndex + 1
-        final JsonNode result = getDataFromPath("", taskIndex + 1, false, 0);
-        final Iterator<JsonNode> iterator = ZendeskUtils.getListRecords(result, task.getTarget().getJsonName());
+        long startTime = 0;
+        long endTime = 0;
+
+        if (Target.SATISFACTION_RATINGS.equals(task.getTarget())){
+            if (getStartDate && task.getStartTime().isPresent()) {
+                startTime = ZendeskDateUtils.getStartTime(task.getStartTime().get());
+            }
+
+            if (task.getEndTime().isPresent()) {
+                endTime = ZendeskDateUtils.isoToEpochSecond(task.getEndTime().get());
+            }
+
+            int i = 1;
+            final Set<String> knownIds = ConcurrentHashMap.newKeySet();
+            while (true) {
+                final JsonNode result = getDataFromPath("", taskIndex +i, false, startTime, endTime);
+                final Iterator<JsonNode> iterator = ZendeskUtils.getListRecords(result, task.getTarget().getJsonName());
+                while (iterator.hasNext()) {
+                    final JsonNode recordJsonNode = iterator.next();
+
+                    if (task.getDedup()) {
+                        final String recordID = recordJsonNode.get(ZendeskConstants.Field.ID).asText();
+
+                        // add success -> no duplicate
+                        if (!knownIds.add(recordID)) {
+                            continue;
+                        }
+                    }
+                    fetchSubResourceAndAddToImporter(recordJsonNode, task, recordImporter);
+
+                    if (Exec.isPreview()) {
+                        break;
+                    }
+                }
+
+                    i++;
+                if(result.has(ZendeskConstants.Field.NEXT_PAGE)){
+                    String next_page = result.get(ZendeskConstants.Field.NEXT_PAGE).textValue();
+                    if (next_page == null){
+                        logger.info("No next page exists. Exiting...");
+                        return;
+                    }
+                    logger.info("Next page = {}", next_page);
+                }
+                }
+
+        }
+        else{
 
-        while (iterator.hasNext()) {
-            fetchSubResourceAndAddToImporter(iterator.next(), task, recordImporter);
+            // Page start from 1 => page = taskIndex + 1
+            final JsonNode result = getDataFromPath("", taskIndex + 1 , false, 0);
+            final Iterator<JsonNode> iterator = ZendeskUtils.getListRecords(result, task.getTarget().getJsonName());
+            while (iterator.hasNext()) {
+                fetchSubResourceAndAddToImporter(iterator.next(), task, recordImporter);
 
-            if (Exec.isPreview()) {
-                break;
+                if (Exec.isPreview()) {
+                    break;
+                }
             }
         }
+
     }
 }

src/main/java/org/embulk/input/zendesk/services/ZendeskSupportAPIService.java

@@ -15,11 +15,20 @@ public ZendeskSupportAPIService(final PluginTask task)
 
     public boolean isSupportIncremental()
     {
-        return !(task.getTarget().equals(Target.TICKET_FORMS) || task.getTarget().equals(Target.TICKET_FIELDS));
+        return !(task.getTarget().equals(Target.TICKET_FORMS)
+                || task.getTarget().equals(Target.SLA_POLICIES)
+                || task.getTarget().equals(Target.TICKET_FIELDS)
+                || task.getTarget().equals(Target.SATISFACTION_RATINGS)
+                || task.getTarget().equals(Target.GROUPS));
     }
 
     @Override
-    protected String buildURI(final int page, long startTime)
+    protected String buildURI(final int page, long startTime){
+        return buildURI(page, startTime);
+    }
+
+    @Override
+    protected String buildURI(final int page, long startTime, long endTime)
     {
         final URIBuilder uriBuilder = ZendeskUtils.getURIBuilder(task.getLoginUrl()).setPath(buildPath());
 
@@ -30,23 +39,45 @@ protected String buildURI(final int page, long startTime)
             }
         }
         else {
-            uriBuilder.setParameter("sort_by", "id")
-                    .setParameter("per_page", String.valueOf(100))
-                    .setParameter("page", String.valueOf(page));
+            if (Target.SATISFACTION_RATINGS.equals(task.getTarget())){
+                if (endTime>0){
+                    uriBuilder.setParameter(ZendeskConstants.Field.START_TIME, String.valueOf(startTime))
+                            .setParameter(ZendeskConstants.Field.END_TIME, String.valueOf(endTime))
+                            .setParameter("sort_by", "id")
+                            .setParameter("per_page", String.valueOf(100))
+                            .setParameter("page", String.valueOf(page));
+                }else{
+                    uriBuilder.setParameter(ZendeskConstants.Field.START_TIME, String.valueOf(startTime))
+                            .setParameter("sort_by", "id")
+                            .setParameter("per_page", String.valueOf(100))
+                            .setParameter("page", String.valueOf(page));
+                }
+            }
+            else{
+                uriBuilder.setParameter("sort_by", "id")
+                        .setParameter("per_page", String.valueOf(100))
+                        .setParameter("page", String.valueOf(page));
+            }
         }
 
         return uriBuilder.toString();
     }
 
     private String buildPath()
     {
-        return (isSupportIncremental()
-                ? ZendeskConstants.Url.API_INCREMENTAL
-                : ZendeskConstants.Url.API) +
-                "/" +
-                (Target.TICKET_METRICS.equals(task.getTarget())
-                        ? Target.TICKETS.toString()
-                        : task.getTarget().toString())
-                + ".json";
+        if(Target.SLA_POLICIES.equals(task.getTarget()))
+        {
+            return ZendeskConstants.Url.API+"/slas/policies";
+        }
+        else{
+            return (isSupportIncremental() && !(Target.SATISFACTION_RATINGS.equals(task.getTarget()))
+                    ? ZendeskConstants.Url.API_INCREMENTAL
+                    : ZendeskConstants.Url.API) +
+                    "/" +
+                    (Target.TICKET_METRICS.equals(task.getTarget())
+                            ? Target.TICKETS.toString()
+                            : task.getTarget().toString())
+                    + ".json";
+        }
     }
 }

src/main/java/org/embulk/input/zendesk/utils/ZendeskConstants.java

@@ -28,6 +28,7 @@ public static class Field
         public static final String GENERATED_TIMESTAMP = "generated_timestamp";
         public static final String UPDATED_AT = "updated_at";
         public static final String ID = "id";
+        public static final String NEXT_PAGE="next_page";
     }
 
     public static class Url

src/main/java/org/embulk/input/zendesk/utils/ZendeskUtils.java

@@ -15,10 +15,13 @@
 import java.net.URISyntaxException;
 import java.util.Base64;
 import java.util.Iterator;
+import org.embulk.spi.Exec;
+import org.slf4j.Logger;
 
 public class ZendeskUtils
 {
     private static final ObjectMapper mapper = new ObjectMapper();
+    private static final Logger logger = Exec.getLogger(ZendeskUtils.class);
 
     static {
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);