<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zeppelin.notebook;

import java.io.IOException;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.zeppelin.common.JsonSerializable;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.GUI;
import org.apache.zeppelin.display.Input;
import org.apache.zeppelin.helium.HeliumPackage;
import org.apache.zeppelin.interpreter.Constants;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.Interpreter.FormType;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterException;
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterInfo;
import org.apache.zeppelin.interpreter.InterpreterNotFoundException;
import org.apache.zeppelin.interpreter.InterpreterOutput;
import org.apache.zeppelin.interpreter.InterpreterOutputListener;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
import org.apache.zeppelin.interpreter.InterpreterResultMessage;
import org.apache.zeppelin.interpreter.InterpreterResultMessageOutput;
import org.apache.zeppelin.interpreter.InterpreterSetting;
import org.apache.zeppelin.interpreter.InterpreterSettingManager;
import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.resource.ResourcePool;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.JobListener;
import org.apache.zeppelin.scheduler.JobWithProgressPoller;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.apache.zeppelin.user.Credentials;
import org.apache.zeppelin.user.UserCredentials;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;

/**
 * Paragraph is a representation of an execution unit.
 */
public class Paragraph extends JobWithProgressPoller<InterpreterResult> implements Cloneable,
    JsonSerializable {

  private static Logger LOGGER = LoggerFactory.getLogger(Paragraph.class);
  private static Pattern REPL_PATTERN =
      Pattern.compile("(\\s*)%([\\w\\.]+)(\\(.*?\\))?.*", Pattern.DOTALL);

  private String title;
  // text is composed of intpText and scriptText.
  private String text;
  private String user;
  private Date dateUpdated;
  // paragraph configs like isOpen, colWidth, etc
  private Map<String, Object> config = new HashMap<>();
  // form and parameter settings
  public GUI settings = new GUI();
  private InterpreterResult results;
  // Application states in this paragraph
  private final List<ApplicationState> apps = new LinkedList<>();

  /************** Transient fields which are not serializabled  into note json **************/
  private transient String intpText;
  private transient Boolean configSettingNeedUpdate = true;
  private transient String scriptText;
  private transient Interpreter interpreter;
  private transient Note note;
  private transient AuthenticationInfo subject;
  // personalized
  private transient Map<String, Paragraph> userParagraphMap = new HashMap<>();
  private transient Map<String, String> localProperties = new HashMap<>();
  private transient Map<String, ParagraphRuntimeInfo> runtimeInfos = new HashMap<>();

  public static String  PARAGRAPH_CONFIG_RUNONSELECTIONCHANGE = "runOnSelectionChange";
  private static boolean PARAGRAPH_CONFIG_RUNONSELECTIONCHANGE_DEFAULT = true;
  public static String  PARAGRAPH_CONFIG_TITLE = "title";
  private static boolean PARAGRAPH_CONFIG_TITLE_DEFAULT = false;
  public static String  PARAGRAPH_CONFIG_CHECK_EMTPY = "checkEmpty";
  private static boolean PARAGRAPH_CONFIG_CHECK_EMTPY_DEFAULT = true;

  @VisibleForTesting
  Paragraph() {
    super(generateId(), null);
  }

  public Paragraph(String paragraphId, Note note, JobListener listener) {
    super(paragraphId, generateId(), listener);
    this.note = note;
  }

  public Paragraph(Note note, JobListener listener) {
    super(generateId(), listener);
    this.note = note;
  }

  // used for clone paragraph
  public Paragraph(Paragraph p2) {
    super(p2.getId(), null);
    this.note = p2.note;
    this.settings.setParams(Maps.newHashMap(p2.settings.getParams()));
    this.settings.setForms(Maps.newLinkedHashMap(p2.settings.getForms()));
    this.setConfig(Maps.newHashMap(p2.config));
    this.setAuthenticationInfo(p2.getAuthenticationInfo());
    this.title = p2.title;
    this.text = p2.text;
    this.results = p2.results;
    setStatus(p2.getStatus());
  }

  private static String generateId() {
    return "paragraph_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt();
  }

  public Map<String, Paragraph> getUserParagraphMap() {
    return userParagraphMap;
  }

  public Paragraph getUserParagraph(String user) {
    if (!userParagraphMap.containsKey(user)) {
      cloneParagraphForUser(user);
    }
    return userParagraphMap.get(user);
  }

  @Override
  public void setResult(InterpreterResult result) {
    this.results = result;
  }

  public Paragraph cloneParagraphForUser(String user) {
    Paragraph p = new Paragraph(this);
    // reset status to READY when clone Paragraph for personalization.
    p.status = Status.READY;
    addUser(p, user);
    return p;
  }

  private void setIntpText(String newIntptext) {
    if (null == intpText || !this.intpText.equals(newIntptext)) {
      this.configSettingNeedUpdate = true;
    }
    this.intpText = newIntptext;
  }

  public void clearUserParagraphs() {
    userParagraphMap.clear();
  }

  public void addUser(Paragraph p, String user) {
    userParagraphMap.put(user, p);
  }

  public String getUser() {
    return user;
  }

  public String getText() {
    return text;
  }

  public void setText(String newText) {
    this.text = newText;
    this.dateUpdated = new Date();
    parseText();
  }

  public void parseText() {
    // parse text to get interpreter component
    if (this.text != null) {
      // clean localProperties, otherwise previous localProperties will be used for the next run
      this.localProperties.clear();
      Matcher matcher = REPL_PATTERN.matcher(this.text);
      if (matcher.matches()) {
        String headingSpace = matcher.group(1);
        setIntpText(matcher.group(2));

        if (matcher.groupCount() == 3 && matcher.group(3) != null) {
          String localPropertiesText = matcher.group(3);
          String[] splits = localPropertiesText.substring(1, localPropertiesText.length() -1)
              .split(",");
          for (String split : splits) {
            String[] kv = split.split("=");
            if (StringUtils.isBlank(split) || kv.length == 0) {
              continue;
            }
            if (kv.length > 2) {
              throw new RuntimeException("Invalid paragraph properties format: " + split);
            }
            if (kv.length == 1) {
              localProperties.put(kv[0].trim(), kv[0].trim());
            } else {
              localProperties.put(kv[0].trim(), kv[1].trim());
            }
          }
          this.scriptText = this.text.substring(headingSpace.length() + intpText.length() +
              localPropertiesText.length() + 1).trim();
        } else {
          this.scriptText = this.text.substring(headingSpace.length() + intpText.length() + 1).trim();
        }
        config.putAll(localProperties);
      } else {
        setIntpText("");
        this.scriptText = this.text.trim();
      }
    }
  }

  public AuthenticationInfo getAuthenticationInfo() {
    return subject;
  }

  public void setAuthenticationInfo(AuthenticationInfo subject) {
    this.subject = subject;
    if (subject != null) {
      this.user = subject.getUser();
    }
  }

  public String getTitle() {
    return title;
  }

  public void setTitle(String title) {
    this.title = title;
  }

  public String getIntpText() {
    return intpText;
  }

  public String getScriptText() {
    return scriptText;
  }

  public void setNote(Note note) {
    this.note = note;
  }

  public Note getNote() {
    return note;
  }

  public Map<String, String> getLocalProperties() {
    return localProperties;
  }

  public boolean isEnabled() {
    Boolean enabled = (Boolean) config.get("enabled");
    return enabled == null || enabled.booleanValue();
  }

  public Interpreter getBindedInterpreter() throws InterpreterNotFoundException {
    return this.note.getInterpreterFactory().getInterpreter(user, note.getId(), intpText,
        note.getDefaultInterpreterGroup());
  }

  public void setInterpreter(Interpreter interpreter) {
    this.interpreter = interpreter;
  }

  public List<InterpreterCompletion> completion(String buffer, int cursor) {
    setText(buffer);
    try {
      this.interpreter = getBindedInterpreter();
    } catch (InterpreterNotFoundException e) {
      LOGGER.debug("Unable to get completion because there's no interpreter bind to it", e);
      return new ArrayList<>();
    }
    cursor = calculateCursorPosition(buffer, cursor);
    InterpreterContext interpreterContext = getInterpreterContext(null);

    try {
      return this.interpreter.completion(this.scriptText, cursor, interpreterContext);
    } catch (InterpreterException e) {
      LOGGER.warn("Fail to get completion", e);
      return new ArrayList<>();
    }
  }

  public int calculateCursorPosition(String buffer, int cursor) {
    if (this.scriptText.isEmpty()) {
      return 0;
    }
    int countCharactersBeforeScript = buffer.indexOf(this.scriptText);
    if (countCharactersBeforeScript > 0) {
      cursor -= countCharactersBeforeScript;
    }

    return cursor;
  }

  @Override
  public InterpreterResult getReturn() {
    return results;
  }

  @Override
  public int progress() {
    try {
      if (this.interpreter != null) {
        return this.interpreter.getProgress(getInterpreterContext(null));
      } else {
        return 0;
      }
    } catch (InterpreterException e) {
      throw new RuntimeException("Fail to get progress", e);
    }
  }

  @Override
  public Map<String, Object> info() {
    return null;
  }

  public boolean shouldSkipRunParagraph() {
    // Because the user can arbitrarily specify the paragraph's interpreter
    // So need to determine the configuration of the interpreter
    // and the secondary interpreter at runtime.
    Map<String, Object> intpConfig = this.config;

    if (!StringUtils.isBlank(intpText)) {
      String[] intpList = intpText.split("\\.");
      if (intpList.length > 0) {
        InterpreterSettingManager intpSettingManager = note.getInterpreterSettingManager();
        String intpName = intpList[0];
        try {
          InterpreterSetting intpSetting = intpSettingManager.getInterpreterSettingByName(intpName);
          String intpInfoName = intpName; // e.g %sh

          // e.g %sh.terminal
          if (intpList.length == 2) {
            intpInfoName = intpList[1];
          }
          InterpreterInfo interpreterInfo = intpSetting.getInterpreterInfo(intpInfoName);
          if (null != interpreterInfo && null != interpreterInfo.getConfig()) {
            intpConfig = interpreterInfo.getConfig();
          }
        } catch (RuntimeException e) {
          LOGGER.error(e.getMessage(), e);
        }
      }
    }

    // check interpreter-setting.json `config.checkEmpty` is equal false
    Object configCheckEmpty = intpConfig.get(PARAGRAPH_CONFIG_CHECK_EMTPY);
    if (null != configCheckEmpty) {
      boolean checkEmtpy = PARAGRAPH_CONFIG_CHECK_EMTPY_DEFAULT;
      try {
        checkEmtpy = (boolean) configCheckEmpty;
      } catch (ClassCastException e) {
        LOGGER.error(e.getMessage(), e);
      } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
      }
      if (!checkEmtpy) {
        LOGGER.info("This interpreter config `interpreter-setting.json` set config.{} = false", 
            PARAGRAPH_CONFIG_CHECK_EMTPY);
        return false;
      }
    }

    return Strings.isNullOrEmpty(scriptText);
  }

  public boolean execute(boolean blocking) {
    if (shouldSkipRunParagraph()) {
      LOGGER.info("Skip to run blank paragraph. {}", getId());
      setStatus(Job.Status.FINISHED);
      return true;
    }

    try {
      this.interpreter = getBindedInterpreter();
      setStatus(Status.READY);

      if (getConfig().get("enabled") == null || (Boolean) getConfig().get("enabled")) {
        setAuthenticationInfo(getAuthenticationInfo());
        interpreter.getScheduler().submit(this);
      }

      if (blocking) {
        while (!getStatus().isCompleted()) {
          try {
            Thread.sleep(100);
          } catch (InterruptedException e) {
            throw new RuntimeException(e);
          }
        }
        return getStatus() == Status.FINISHED;
      } else {
        return true;
      }
    } catch (InterpreterNotFoundException e) {
      InterpreterResult intpResult =
          new InterpreterResult(InterpreterResult.Code.ERROR,
                  String.format("Interpreter %s not found", this.intpText));
      setReturn(intpResult, e);
      setStatus(Job.Status.ERROR);
      return false;
    }
  }

  @Override
  protected InterpreterResult jobRun() throws Throwable {
    try {
      this.runtimeInfos.clear();
      this.interpreter = getBindedInterpreter();
      if (this.interpreter == null) {
        LOGGER.error("Can not find interpreter name " + intpText);
        throw new RuntimeException("Can not find interpreter for " + intpText);
      }
      LOGGER.info("Run paragraph [paragraph_id: {}, interpreter: {}, note_id: {}, user: {}]",
              getId(), this.interpreter.getClassName(), note.getId(), subject.getUser());
      InterpreterSetting interpreterSetting = ((ManagedInterpreterGroup)
              interpreter.getInterpreterGroup()).getInterpreterSetting();
      if (interpreterSetting != null) {
        interpreterSetting.waitForReady();
      }
      if (this.user != null) {
        if (subject != null && !interpreterSetting.isUserAuthorized(subject.getUsersAndRoles())) {
          String msg = String.format("%s has no permission for %s", subject.getUser(), intpText);
          LOGGER.error(msg);
          return new InterpreterResult(Code.ERROR, msg);
        }
      }

      for (Paragraph p : userParagraphMap.values()) {
        p.setText(getText());
      }

      // inject form
      String script = this.scriptText;
      if ("simple".equalsIgnoreCase(localProperties.get("form")) ||
              interpreter.getFormType() == FormType.SIMPLE) {
        // inputs will be built from script body
        LinkedHashMap<String, Input> inputs = Input.extractSimpleQueryForm(script, false);
        LinkedHashMap<String, Input> noteInputs = Input.extractSimpleQueryForm(script, true);
        final AngularObjectRegistry angularRegistry =
                interpreter.getInterpreterGroup().getAngularObjectRegistry();
        String scriptBody = extractVariablesFromAngularRegistry(script, inputs, angularRegistry);

        settings.setForms(inputs);
        if (!noteInputs.isEmpty()) {
          if (!note.getNoteForms().isEmpty()) {
            Map<String, Input> currentNoteForms = note.getNoteForms();
            for (String s : noteInputs.keySet()) {
              if (!currentNoteForms.containsKey(s)) {
                currentNoteForms.put(s, noteInputs.get(s));
              }
            }
          } else {
            note.setNoteForms(noteInputs);
          }
        }
        script = Input.getSimpleQuery(note.getNoteParams(), scriptBody, true);
        script = Input.getSimpleQuery(settings.getParams(), script, false);
      } else {
        settings.clear();
      }

      LOGGER.debug("RUN : " + script);
      try {
        InterpreterContext context = getInterpreterContext();
        InterpreterContext.set(context);

        // Inject credentials
        String injectPropStr = interpreter.getProperty(Constants.INJECT_CREDENTIALS, "false");
        injectPropStr = context.getStringLocalProperty(Constants.INJECT_CREDENTIALS, injectPropStr);
        boolean shouldInjectCredentials = Boolean.parseBoolean(injectPropStr);

        InterpreterResult ret = null;
        if (shouldInjectCredentials) {
          UserCredentials creds = context.getAuthenticationInfo().getUserCredentials();
          CredentialInjector credinjector = new CredentialInjector(creds);
          String code = credinjector.replaceCredentials(script);
          ret = interpreter.interpret(code, context);
          ret = credinjector.hidePasswords(ret);
        } else {
          ret = interpreter.interpret(script, context);
        }

        if (interpreter.getFormType() == FormType.NATIVE) {
          note.setNoteParams(context.getNoteGui().getParams());
          note.setNoteForms(context.getNoteGui().getForms());
        }

        if (Code.KEEP_PREVIOUS_RESULT == ret.code()) {
          return getReturn();
        }

        context.out.flush();
        List<InterpreterResultMessage> resultMessages = context.out.toInterpreterResultMessage();
        resultMessages.addAll(ret.message());
        InterpreterResult res = new InterpreterResult(ret.code(), resultMessages);
        Paragraph p = getUserParagraph(getUser());
        if (null != p) {
          p.setResult(res);
          p.settings.setParams(settings.getParams());
        }

        // After the paragraph is executed,
        // need to apply the paragraph to the configuration in the
        // `interpreter-setting.json` config
        if (this.configSettingNeedUpdate) {
          this.configSettingNeedUpdate = false;
          InterpreterSettingManager intpSettingManager
                  = this.note.getInterpreterSettingManager();
          if (null != intpSettingManager) {
            InterpreterGroup intpGroup = interpreter.getInterpreterGroup();
            if (null != intpGroup && intpGroup instanceof ManagedInterpreterGroup) {
              String name = ((ManagedInterpreterGroup) intpGroup).getInterpreterSetting().getName();
              Map<String, Object> config
                      = intpSettingManager.getConfigSetting(name);
              mergeConfig(config);
            }
          }
        }

        return res;
      } finally {
        InterpreterContext.remove();
      }
    } catch (Exception e) {
      return new InterpreterResult(Code.ERROR, ExceptionUtils.getStackTrace(e));
    }
  }

  @Override
  protected boolean jobAbort() {
    if (interpreter == null) {
      return true;
    }
    try {
      interpreter.cancel(getInterpreterContext(null));
    } catch (InterpreterException e) {
      throw new RuntimeException(e);
    }<commit_msg>[hotfix] error message is duplicated<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zeppelin.notebook;

import java.io.IOException;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.zeppelin.common.JsonSerializable;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.GUI;
import org.apache.zeppelin.display.Input;
import org.apache.zeppelin.helium.HeliumPackage;
import org.apache.zeppelin.interpreter.Constants;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.Interpreter.FormType;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterException;
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterInfo;
import org.apache.zeppelin.interpreter.InterpreterNotFoundException;
import org.apache.zeppelin.interpreter.InterpreterOutput;
import org.apache.zeppelin.interpreter.InterpreterOutputListener;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
import org.apache.zeppelin.interpreter.InterpreterResultMessage;
import org.apache.zeppelin.interpreter.InterpreterResultMessageOutput;
import org.apache.zeppelin.interpreter.InterpreterSetting;
import org.apache.zeppelin.interpreter.InterpreterSettingManager;
import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.resource.ResourcePool;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.JobListener;
import org.apache.zeppelin.scheduler.JobWithProgressPoller;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.apache.zeppelin.user.Credentials;
import org.apache.zeppelin.user.UserCredentials;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;

/**
 * Paragraph is a representation of an execution unit.
 */
public class Paragraph extends JobWithProgressPoller<InterpreterResult> implements Cloneable,
    JsonSerializable {

  private static Logger LOGGER = LoggerFactory.getLogger(Paragraph.class);
  private static Pattern REPL_PATTERN =
      Pattern.compile("(\\s*)%([\\w\\.]+)(\\(.*?\\))?.*", Pattern.DOTALL);

  private String title;
  // text is composed of intpText and scriptText.
  private String text;
  private String user;
  private Date dateUpdated;
  // paragraph configs like isOpen, colWidth, etc
  private Map<String, Object> config = new HashMap<>();
  // form and parameter settings
  public GUI settings = new GUI();
  private InterpreterResult results;
  // Application states in this paragraph
  private final List<ApplicationState> apps = new LinkedList<>();

  /************** Transient fields which are not serializabled  into note json **************/
  private transient String intpText;
  private transient Boolean configSettingNeedUpdate = true;
  private transient String scriptText;
  private transient Interpreter interpreter;
  private transient Note note;
  private transient AuthenticationInfo subject;
  // personalized
  private transient Map<String, Paragraph> userParagraphMap = new HashMap<>();
  private transient Map<String, String> localProperties = new HashMap<>();
  private transient Map<String, ParagraphRuntimeInfo> runtimeInfos = new HashMap<>();

  public static String  PARAGRAPH_CONFIG_RUNONSELECTIONCHANGE = "runOnSelectionChange";
  private static boolean PARAGRAPH_CONFIG_RUNONSELECTIONCHANGE_DEFAULT = true;
  public static String  PARAGRAPH_CONFIG_TITLE = "title";
  private static boolean PARAGRAPH_CONFIG_TITLE_DEFAULT = false;
  public static String  PARAGRAPH_CONFIG_CHECK_EMTPY = "checkEmpty";
  private static boolean PARAGRAPH_CONFIG_CHECK_EMTPY_DEFAULT = true;

  @VisibleForTesting
  Paragraph() {
    super(generateId(), null);
  }

  public Paragraph(String paragraphId, Note note, JobListener listener) {
    super(paragraphId, generateId(), listener);
    this.note = note;
  }

  public Paragraph(Note note, JobListener listener) {
    super(generateId(), listener);
    this.note = note;
  }

  // used for clone paragraph
  public Paragraph(Paragraph p2) {
    super(p2.getId(), null);
    this.note = p2.note;
    this.settings.setParams(Maps.newHashMap(p2.settings.getParams()));
    this.settings.setForms(Maps.newLinkedHashMap(p2.settings.getForms()));
    this.setConfig(Maps.newHashMap(p2.config));
    this.setAuthenticationInfo(p2.getAuthenticationInfo());
    this.title = p2.title;
    this.text = p2.text;
    this.results = p2.results;
    setStatus(p2.getStatus());
  }

  private static String generateId() {
    return "paragraph_" + System.currentTimeMillis() + "_" + new SecureRandom().nextInt();
  }

  public Map<String, Paragraph> getUserParagraphMap() {
    return userParagraphMap;
  }

  public Paragraph getUserParagraph(String user) {
    if (!userParagraphMap.containsKey(user)) {
      cloneParagraphForUser(user);
    }
    return userParagraphMap.get(user);
  }

  @Override
  public void setResult(InterpreterResult result) {
    this.results = result;
  }

  public Paragraph cloneParagraphForUser(String user) {
    Paragraph p = new Paragraph(this);
    // reset status to READY when clone Paragraph for personalization.
    p.status = Status.READY;
    addUser(p, user);
    return p;
  }

  private void setIntpText(String newIntptext) {
    if (null == intpText || !this.intpText.equals(newIntptext)) {
      this.configSettingNeedUpdate = true;
    }
    this.intpText = newIntptext;
  }

  public void clearUserParagraphs() {
    userParagraphMap.clear();
  }

  public void addUser(Paragraph p, String user) {
    userParagraphMap.put(user, p);
  }

  public String getUser() {
    return user;
  }

  public String getText() {
    return text;
  }

  public void setText(String newText) {
    this.text = newText;
    this.dateUpdated = new Date();
    parseText();
  }

  public void parseText() {
    // parse text to get interpreter component
    if (this.text != null) {
      // clean localProperties, otherwise previous localProperties will be used for the next run
      this.localProperties.clear();
      Matcher matcher = REPL_PATTERN.matcher(this.text);
      if (matcher.matches()) {
        String headingSpace = matcher.group(1);
        setIntpText(matcher.group(2));

        if (matcher.groupCount() == 3 && matcher.group(3) != null) {
          String localPropertiesText = matcher.group(3);
          String[] splits = localPropertiesText.substring(1, localPropertiesText.length() -1)
              .split(",");
          for (String split : splits) {
            String[] kv = split.split("=");
            if (StringUtils.isBlank(split) || kv.length == 0) {
              continue;
            }
            if (kv.length > 2) {
              throw new RuntimeException("Invalid paragraph properties format: " + split);
            }
            if (kv.length == 1) {
              localProperties.put(kv[0].trim(), kv[0].trim());
            } else {
              localProperties.put(kv[0].trim(), kv[1].trim());
            }
          }
          this.scriptText = this.text.substring(headingSpace.length() + intpText.length() +
              localPropertiesText.length() + 1).trim();
        } else {
          this.scriptText = this.text.substring(headingSpace.length() + intpText.length() + 1).trim();
        }
        config.putAll(localProperties);
      } else {
        setIntpText("");
        this.scriptText = this.text.trim();
      }
    }
  }

  public AuthenticationInfo getAuthenticationInfo() {
    return subject;
  }

  public void setAuthenticationInfo(AuthenticationInfo subject) {
    this.subject = subject;
    if (subject != null) {
      this.user = subject.getUser();
    }
  }

  public String getTitle() {
    return title;
  }

  public void setTitle(String title) {
    this.title = title;
  }

  public String getIntpText() {
    return intpText;
  }

  public String getScriptText() {
    return scriptText;
  }

  public void setNote(Note note) {
    this.note = note;
  }

  public Note getNote() {
    return note;
  }

  public Map<String, String> getLocalProperties() {
    return localProperties;
  }

  public boolean isEnabled() {
    Boolean enabled = (Boolean) config.get("enabled");
    return enabled == null || enabled.booleanValue();
  }

  public Interpreter getBindedInterpreter() throws InterpreterNotFoundException {
    return this.note.getInterpreterFactory().getInterpreter(user, note.getId(), intpText,
        note.getDefaultInterpreterGroup());
  }

  public void setInterpreter(Interpreter interpreter) {
    this.interpreter = interpreter;
  }

  public List<InterpreterCompletion> completion(String buffer, int cursor) {
    setText(buffer);
    try {
      this.interpreter = getBindedInterpreter();
    } catch (InterpreterNotFoundException e) {
      LOGGER.debug("Unable to get completion because there's no interpreter bind to it", e);
      return new ArrayList<>();
    }
    cursor = calculateCursorPosition(buffer, cursor);
    InterpreterContext interpreterContext = getInterpreterContext(null);

    try {
      return this.interpreter.completion(this.scriptText, cursor, interpreterContext);
    } catch (InterpreterException e) {
      LOGGER.warn("Fail to get completion", e);
      return new ArrayList<>();
    }
  }

  public int calculateCursorPosition(String buffer, int cursor) {
    if (this.scriptText.isEmpty()) {
      return 0;
    }
    int countCharactersBeforeScript = buffer.indexOf(this.scriptText);
    if (countCharactersBeforeScript > 0) {
      cursor -= countCharactersBeforeScript;
    }

    return cursor;
  }

  @Override
  public InterpreterResult getReturn() {
    return results;
  }

  @Override
  public int progress() {
    try {
      if (this.interpreter != null) {
        return this.interpreter.getProgress(getInterpreterContext(null));
      } else {
        return 0;
      }
    } catch (InterpreterException e) {
      throw new RuntimeException("Fail to get progress", e);
    }
  }

  @Override
  public Map<String, Object> info() {
    return null;
  }

  public boolean shouldSkipRunParagraph() {
    // Because the user can arbitrarily specify the paragraph's interpreter
    // So need to determine the configuration of the interpreter
    // and the secondary interpreter at runtime.
    Map<String, Object> intpConfig = this.config;

    if (!StringUtils.isBlank(intpText)) {
      String[] intpList = intpText.split("\\.");
      if (intpList.length > 0) {
        InterpreterSettingManager intpSettingManager = note.getInterpreterSettingManager();
        String intpName = intpList[0];
        try {
          InterpreterSetting intpSetting = intpSettingManager.getInterpreterSettingByName(intpName);
          String intpInfoName = intpName; // e.g %sh

          // e.g %sh.terminal
          if (intpList.length == 2) {
            intpInfoName = intpList[1];
          }
          InterpreterInfo interpreterInfo = intpSetting.getInterpreterInfo(intpInfoName);
          if (null != interpreterInfo && null != interpreterInfo.getConfig()) {
            intpConfig = interpreterInfo.getConfig();
          }
        } catch (RuntimeException e) {
          LOGGER.error(e.getMessage(), e);
        }
      }
    }

    // check interpreter-setting.json `config.checkEmpty` is equal false
    Object configCheckEmpty = intpConfig.get(PARAGRAPH_CONFIG_CHECK_EMTPY);
    if (null != configCheckEmpty) {
      boolean checkEmtpy = PARAGRAPH_CONFIG_CHECK_EMTPY_DEFAULT;
      try {
        checkEmtpy = (boolean) configCheckEmpty;
      } catch (ClassCastException e) {
        LOGGER.error(e.getMessage(), e);
      } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
      }
      if (!checkEmtpy) {
        LOGGER.info("This interpreter config `interpreter-setting.json` set config.{} = false", 
            PARAGRAPH_CONFIG_CHECK_EMTPY);
        return false;
      }
    }

    return Strings.isNullOrEmpty(scriptText);
  }

  public boolean execute(boolean blocking) {
    if (shouldSkipRunParagraph()) {
      LOGGER.info("Skip to run blank paragraph. {}", getId());
      setStatus(Job.Status.FINISHED);
      return true;
    }

    try {
      this.interpreter = getBindedInterpreter();
      setStatus(Status.READY);

      if (getConfig().get("enabled") == null || (Boolean) getConfig().get("enabled")) {
        setAuthenticationInfo(getAuthenticationInfo());
        interpreter.getScheduler().submit(this);
      }

      if (blocking) {
        while (!getStatus().isCompleted()) {
          try {
            Thread.sleep(100);
          } catch (InterruptedException e) {
            throw new RuntimeException(e);
          }
        }
        return getStatus() == Status.FINISHED;
      } else {
        return true;
      }
    } catch (InterpreterNotFoundException e) {
      InterpreterResult intpResult =
          new InterpreterResult(InterpreterResult.Code.ERROR,
                  String.format("Interpreter %s not found", this.intpText));
      setReturn(intpResult, e);
      setStatus(Job.Status.ERROR);
      return false;
    }
  }

  @Override
  protected InterpreterResult jobRun() throws Throwable {
    try {
      this.runtimeInfos.clear();
      this.interpreter = getBindedInterpreter();
      if (this.interpreter == null) {
        LOGGER.error("Can not find interpreter name " + intpText);
        throw new RuntimeException("Can not find interpreter for " + intpText);
      }
      LOGGER.info("Run paragraph [paragraph_id: {}, interpreter: {}, note_id: {}, user: {}]",
              getId(), this.interpreter.getClassName(), note.getId(), subject.getUser());
      InterpreterSetting interpreterSetting = ((ManagedInterpreterGroup)
              interpreter.getInterpreterGroup()).getInterpreterSetting();
      if (interpreterSetting != null) {
        interpreterSetting.waitForReady();
      }
      if (this.user != null) {
        if (subject != null && !interpreterSetting.isUserAuthorized(subject.getUsersAndRoles())) {
          String msg = String.format("%s has no permission for %s", subject.getUser(), intpText);
          LOGGER.error(msg);
          return new InterpreterResult(Code.ERROR, msg);
        }
      }

      for (Paragraph p : userParagraphMap.values()) {
        p.setText(getText());
      }

      // inject form
      String script = this.scriptText;
      if ("simple".equalsIgnoreCase(localProperties.get("form")) ||
              interpreter.getFormType() == FormType.SIMPLE) {
        // inputs will be built from script body
        LinkedHashMap<String, Input> inputs = Input.extractSimpleQueryForm(script, false);
        LinkedHashMap<String, Input> noteInputs = Input.extractSimpleQueryForm(script, true);
        final AngularObjectRegistry angularRegistry =
                interpreter.getInterpreterGroup().getAngularObjectRegistry();
        String scriptBody = extractVariablesFromAngularRegistry(script, inputs, angularRegistry);

        settings.setForms(inputs);
        if (!noteInputs.isEmpty()) {
          if (!note.getNoteForms().isEmpty()) {
            Map<String, Input> currentNoteForms = note.getNoteForms();
            for (String s : noteInputs.keySet()) {
              if (!currentNoteForms.containsKey(s)) {
                currentNoteForms.put(s, noteInputs.get(s));
              }
            }
          } else {
            note.setNoteForms(noteInputs);
          }
        }
        script = Input.getSimpleQuery(note.getNoteParams(), scriptBody, true);
        script = Input.getSimpleQuery(settings.getParams(), script, false);
      } else {
        settings.clear();
      }

      LOGGER.debug("RUN : " + script);
      try {
        InterpreterContext context = getInterpreterContext();
        InterpreterContext.set(context);

        // Inject credentials
        String injectPropStr = interpreter.getProperty(Constants.INJECT_CREDENTIALS, "false");
        injectPropStr = context.getStringLocalProperty(Constants.INJECT_CREDENTIALS, injectPropStr);
        boolean shouldInjectCredentials = Boolean.parseBoolean(injectPropStr);

        InterpreterResult ret = null;
        if (shouldInjectCredentials) {
          UserCredentials creds = context.getAuthenticationInfo().getUserCredentials();
          CredentialInjector credinjector = new CredentialInjector(creds);
          String code = credinjector.replaceCredentials(script);
          ret = interpreter.interpret(code, context);
          ret = credinjector.hidePasswords(ret);
        } else {
          ret = interpreter.interpret(script, context);
        }

        if (interpreter.getFormType() == FormType.NATIVE) {
          note.setNoteParams(context.getNoteGui().getParams());
          note.setNoteForms(context.getNoteGui().getForms());
        }

        if (Code.KEEP_PREVIOUS_RESULT == ret.code()) {
          return getReturn();
        }

        context.out.flush();
        List<InterpreterResultMessage> resultMessages = context.out.toInterpreterResultMessage();
        resultMessages.addAll(ret.message());
        InterpreterResult res = new InterpreterResult(ret.code(), resultMessages);
        Paragraph p = getUserParagraph(getUser());
        if (null != p) {
          p.setResult(res);
          p.settings.setParams(settings.getParams());
        }

        // After the paragraph is executed,
        // need to apply the paragraph to the configuration in the
        // `interpreter-setting.json` config
        if (this.configSettingNeedUpdate) {
          this.configSettingNeedUpdate = false;
          InterpreterSettingManager intpSettingManager
                  = this.note.getInterpreterSettingManager();
          if (null != intpSettingManager) {
            InterpreterGroup intpGroup = interpreter.getInterpreterGroup();
            if (null != intpGroup && intpGroup instanceof ManagedInterpreterGroup) {
              String name = ((ManagedInterpreterGroup) intpGroup).getInterpreterSetting().getName();
              Map<String, Object> config
                      = intpSettingManager.getConfigSetting(name);
              mergeConfig(config);
            }
          }
        }

        return res;
      } finally {
        InterpreterContext.remove();
      }
    } catch (Exception e) {
      return new InterpreterResult(Code.ERROR, e.getMessage());
    }
  }

  @Override
  protected boolean jobAbort() {
    if (interpreter == null) {
      return true;
    }
    try {
      interpreter.cancel(getInterpreterContext(null));
    } catch (InterpreterException e) {
      throw new RuntimeException(e);
    }
{'fraction_non_alphanumeric': 0.10619557103757153, 'fraction_numerical': 0.002139835780044787, 'mean_word_length': 3.143829260748531, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3111942', 'n_tokens_mistral': 11267, 'n_tokens_neox': 11296, 'n_words': 2546}