<commit_before>/**
 * (c) 2014 StreamSets, Inc. All rights reserved. May not
 * be copied, modified, or distributed in whole or part without
 * written consent of StreamSets, Inc.
 */
package com.streamsets.pipeline.lib.parser;

import com.streamsets.pipeline.api.Stage;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.config.CsvHeader;
import com.streamsets.pipeline.config.CsvMode;
import com.streamsets.pipeline.config.JsonMode;
import com.streamsets.pipeline.lib.io.OverrunReader;
import com.streamsets.pipeline.lib.parser.delimited.DelimitedCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.json.JsonCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.sdcrecord.JsonSdcRecordCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.text.TextCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.xml.XmlCharDataParserFactory;

import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public abstract class CharDataParserFactory {

  public DataParser getParser(String id, String data) throws DataParserException {
    return getParser(id, new OverrunReader(new StringReader(data), 0, false), 0);
  }

  public DataParser getParser(File file, int overrunLimit, long fileOffset) throws DataParserException {
    FileReader fileReader;
    try {
      fileReader = new FileReader(file);
    } catch (IOException ex) {
      throw new DataParserException(Errors.DATA_PARSER_00, file.getAbsolutePath(), ex.getMessage(), ex);
    }
    try {
      OverrunReader reader = new OverrunReader(fileReader, overrunLimit, false);
      return getParser(file.getName(), reader, fileOffset);
    } catch (DataParserException ex) {
      try {
        fileReader.close();
      } catch (IOException ioEx) {
        //NOP
      }
      throw ex;
    }
  }

  // the reader must be in position zero, the getParser() call will fast forward while initializing the parser
  public abstract DataParser getParser(String id, OverrunReader reader, long readerOffset) throws DataParserException;


  public enum Format {
    TEXT(),
    JSON(JsonMode.class),
    XML(),
    DELIMITED(CsvMode.class, CsvHeader.class),
    SDC_RECORD(),
    LOG(),

    ;

    private final Class<? extends Enum>[] modes;

    Format(Class<? extends Enum> ... modes) {
      this.modes = modes;
    }

    public Class<? extends Enum>[] getModes() {
      return modes;
    }

  }

  public static class Builder {
    private static final Map<String, Object> DEFAULT_CONFIGS;

    static {
      Map<String, Object> defaultConfigs = new HashMap<>();
      TextCharDataParserFactory.registerConfigs(defaultConfigs);
      JsonCharDataParserFactory.registerConfigs(defaultConfigs);
      XmlCharDataParserFactory.registerConfigs(defaultConfigs);
      DelimitedCharDataParserFactory.registerConfigs(defaultConfigs);
      JsonSdcRecordCharDataParserFactory.registerConfigs(defaultConfigs);
      DEFAULT_CONFIGS = Collections.unmodifiableMap(defaultConfigs);
    }

    private Format format;
    private Map<Class<? extends Enum>, Enum> modes;
    private Stage.Context context;
    private int maxDataLen;
    private Map<String, Object> configs;

    public Builder(Stage.Context context, Format format) {
      Utils.checkNotNull(context, "context");
      this.format = Utils.checkNotNull(format, "format");
      this.context = context;
      modes = new HashMap<>();
      configs = new HashMap<>(DEFAULT_CONFIGS);
    }

    public Builder setMode(Enum mode) {
      Utils.checkNotNull(mode, "mode");
      Utils.checkState(format.getModes() != null, Utils.formatL("Format '{}' does not support any mode", format));
      boolean supported = false;
      for (Class<? extends Enum> eMode : format.getModes()) {
        if (eMode.isInstance(mode)) {
          supported = true;
          break;
        }
      }
      Utils.checkState(supported, Utils.formatL("Format '{}' does not support '{}' mode", format,
                                                mode.getClass().getSimpleName()));
      modes.put(mode.getClass(), mode);
      return this;
    }

    public Builder setConfig(String key, Object value) {
      Utils.checkNotNull(key, "key");
      Utils.checkNotNull(value, "value");
      Utils.checkArgument(DEFAULT_CONFIGS.containsKey(key), Utils.formatL("Unsupported configuration '{}'", key));
      Utils.checkArgument(DEFAULT_CONFIGS.get(key).getClass() == value.getClass(),
                          Utils.formatL("Configuration '{}' must be of type '{}'", key,
                                        DEFAULT_CONFIGS.get(key).getClass().getSimpleName()));
      configs.put(key, value);
      return this;
    }

    public Builder setMaxDataLen(int maxDataLen) {
      Utils.checkArgument(maxDataLen > 0 || maxDataLen == -1, Utils.formatL(
          "maxDataLen '{}' cannot be zero, use -1 to disable it", maxDataLen));
      this.maxDataLen = maxDataLen;
      return this;
    }

    public CharDataParserFactory build() {
      Utils.checkState(modes.size() == format.getModes().length, "All required modes have not been set");
      Utils.checkState(maxDataLen != 0, "maxDataLen has not been set");
      return build(context, format, modes, maxDataLen, configs);
    }

    CharDataParserFactory build(Stage.Context context, Format format, Map<Class<? extends Enum>, Enum> modes,
        int maxDataLen, Map<String, Object> configs) {
      CharDataParserFactory factory;
      switch (format) {
        case TEXT:
          factory = new TextCharDataParserFactory(context, maxDataLen, configs);
          break;
        case JSON:
          JsonMode jsonMode = (JsonMode) modes.get(JsonMode.class);
          factory = new JsonCharDataParserFactory(context, maxDataLen, jsonMode.getFormat(), configs);
          break;
        case XML:
          factory = new XmlCharDataParserFactory(context, maxDataLen, configs);
          break;
        case DELIMITED:
          CsvMode csvMode = (CsvMode) modes.get(CsvMode.class);
          CsvHeader csvHeader = (CsvHeader) modes.get(CsvHeader.class);
          factory = new DelimitedCharDataParserFactory(context, csvMode.getFormat(), csvHeader, configs);
          break;
        case SDC_RECORD:
          factory = new JsonSdcRecordCharDataParserFactory(context, maxDataLen, configs);
          break;
        case LOG:
          throw new UnsupportedOperationException();
        default:
          throw new IllegalStateException(Utils.format("Unsupported format '{}'", format));
      }
      return factory;
    }
  }

}<commit_msg>SDC-266. Data parser factory, changed exception on not (yet) implemented LOG format<commit_after>/**
 * (c) 2014 StreamSets, Inc. All rights reserved. May not
 * be copied, modified, or distributed in whole or part without
 * written consent of StreamSets, Inc.
 */
package com.streamsets.pipeline.lib.parser;

import com.streamsets.pipeline.api.Stage;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.config.CsvHeader;
import com.streamsets.pipeline.config.CsvMode;
import com.streamsets.pipeline.config.JsonMode;
import com.streamsets.pipeline.lib.io.OverrunReader;
import com.streamsets.pipeline.lib.parser.delimited.DelimitedCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.json.JsonCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.sdcrecord.JsonSdcRecordCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.text.TextCharDataParserFactory;
import com.streamsets.pipeline.lib.parser.xml.XmlCharDataParserFactory;

import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public abstract class CharDataParserFactory {

  public DataParser getParser(String id, String data) throws DataParserException {
    return getParser(id, new OverrunReader(new StringReader(data), 0, false), 0);
  }

  public DataParser getParser(File file, int overrunLimit, long fileOffset) throws DataParserException {
    FileReader fileReader;
    try {
      fileReader = new FileReader(file);
    } catch (IOException ex) {
      throw new DataParserException(Errors.DATA_PARSER_00, file.getAbsolutePath(), ex.getMessage(), ex);
    }
    try {
      OverrunReader reader = new OverrunReader(fileReader, overrunLimit, false);
      return getParser(file.getName(), reader, fileOffset);
    } catch (DataParserException ex) {
      try {
        fileReader.close();
      } catch (IOException ioEx) {
        //NOP
      }
      throw ex;
    }
  }

  // the reader must be in position zero, the getParser() call will fast forward while initializing the parser
  public abstract DataParser getParser(String id, OverrunReader reader, long readerOffset) throws DataParserException;


  public enum Format {
    TEXT(),
    JSON(JsonMode.class),
    XML(),
    DELIMITED(CsvMode.class, CsvHeader.class),
    SDC_RECORD(),
    LOG(),

    ;

    private final Class<? extends Enum>[] modes;

    Format(Class<? extends Enum> ... modes) {
      this.modes = modes;
    }

    public Class<? extends Enum>[] getModes() {
      return modes;
    }

  }

  public static class Builder {
    private static final Map<String, Object> DEFAULT_CONFIGS;

    static {
      Map<String, Object> defaultConfigs = new HashMap<>();
      TextCharDataParserFactory.registerConfigs(defaultConfigs);
      JsonCharDataParserFactory.registerConfigs(defaultConfigs);
      XmlCharDataParserFactory.registerConfigs(defaultConfigs);
      DelimitedCharDataParserFactory.registerConfigs(defaultConfigs);
      JsonSdcRecordCharDataParserFactory.registerConfigs(defaultConfigs);
      DEFAULT_CONFIGS = Collections.unmodifiableMap(defaultConfigs);
    }

    private Format format;
    private Map<Class<? extends Enum>, Enum> modes;
    private Stage.Context context;
    private int maxDataLen;
    private Map<String, Object> configs;

    public Builder(Stage.Context context, Format format) {
      Utils.checkNotNull(context, "context");
      this.format = Utils.checkNotNull(format, "format");
      this.context = context;
      modes = new HashMap<>();
      configs = new HashMap<>(DEFAULT_CONFIGS);
    }

    public Builder setMode(Enum mode) {
      Utils.checkNotNull(mode, "mode");
      Utils.checkState(format.getModes() != null, Utils.formatL("Format '{}' does not support any mode", format));
      boolean supported = false;
      for (Class<? extends Enum> eMode : format.getModes()) {
        if (eMode.isInstance(mode)) {
          supported = true;
          break;
        }
      }
      Utils.checkState(supported, Utils.formatL("Format '{}' does not support '{}' mode", format,
                                                mode.getClass().getSimpleName()));
      modes.put(mode.getClass(), mode);
      return this;
    }

    public Builder setConfig(String key, Object value) {
      Utils.checkNotNull(key, "key");
      Utils.checkNotNull(value, "value");
      Utils.checkArgument(DEFAULT_CONFIGS.containsKey(key), Utils.formatL("Unsupported configuration '{}'", key));
      Utils.checkArgument(DEFAULT_CONFIGS.get(key).getClass() == value.getClass(),
                          Utils.formatL("Configuration '{}' must be of type '{}'", key,
                                        DEFAULT_CONFIGS.get(key).getClass().getSimpleName()));
      configs.put(key, value);
      return this;
    }

    public Builder setMaxDataLen(int maxDataLen) {
      Utils.checkArgument(maxDataLen > 0 || maxDataLen == -1, Utils.formatL(
          "maxDataLen '{}' cannot be zero, use -1 to disable it", maxDataLen));
      this.maxDataLen = maxDataLen;
      return this;
    }

    public CharDataParserFactory build() {
      Utils.checkState(modes.size() == format.getModes().length, "All required modes have not been set");
      Utils.checkState(maxDataLen != 0, "maxDataLen has not been set");
      return build(context, format, modes, maxDataLen, configs);
    }

    CharDataParserFactory build(Stage.Context context, Format format, Map<Class<? extends Enum>, Enum> modes,
        int maxDataLen, Map<String, Object> configs) {
      CharDataParserFactory factory;
      switch (format) {
        case TEXT:
          factory = new TextCharDataParserFactory(context, maxDataLen, configs);
          break;
        case JSON:
          JsonMode jsonMode = (JsonMode) modes.get(JsonMode.class);
          factory = new JsonCharDataParserFactory(context, maxDataLen, jsonMode.getFormat(), configs);
          break;
        case XML:
          factory = new XmlCharDataParserFactory(context, maxDataLen, configs);
          break;
        case DELIMITED:
          CsvMode csvMode = (CsvMode) modes.get(CsvMode.class);
          CsvHeader csvHeader = (CsvHeader) modes.get(CsvHeader.class);
          factory = new DelimitedCharDataParserFactory(context, csvMode.getFormat(), csvHeader, configs);
          break;
        case SDC_RECORD:
          factory = new JsonSdcRecordCharDataParserFactory(context, maxDataLen, configs);
          break;
        case LOG:
          throw new IllegalArgumentException("LOG not yet implemented");
        default:
          throw new IllegalStateException(Utils.format("Unsupported format '{}'", format));
      }
      return factory;
    }
  }

}
{'fraction_non_alphanumeric': 0.1054456906729634, 'fraction_numerical': 0.0019923258559622196, 'mean_word_length': 3.2902817347261792, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3232495', 'n_tokens_mistral': 3821, 'n_tokens_neox': 3768, 'n_words': 922}