fix CommonParams qupload setting and add path config method from LogPath.

Author: wubingheng

src/main/java/com/qiniu/convert/LineToMap.java

@@ -18,7 +18,7 @@ public LineToMap(String parseType, String separator, String addKeyPrefix, String
             this.lineParser = line -> process(addKeyPrefix, rmKeyPrefix, ConvertingUtils.toPair(line, indexMap, new StringMapPair()));
         } else if ("csv".equals(parseType)) {
             this.lineParser = line -> process(addKeyPrefix, rmKeyPrefix, ConvertingUtils.toPair(line, ",", indexMap, new StringMapPair()));
-        } else if ("tab".equals(parseType) || "self".equals(parseType)) {
+        } else if ("tab".equals(parseType)) {
             this.lineParser = line -> process(addKeyPrefix, rmKeyPrefix, ConvertingUtils.toPair(line, separator, indexMap, new StringMapPair()));
         } else {
             throw new IOException("please check your format for line to map.");

src/main/java/com/qiniu/datasource/CloudStorageContainer.java

@@ -228,7 +228,7 @@ public void export(IStorageLister<E> lister, IResultOutput saver, ILineProcess<T
                 record = json.toString();
                 progressMap.put(lister.getPrefix(), record);
                 try { FileUtils.createIfNotExists(procedureLogFile); } catch (IOException ignored) {}
-                procedureLogger.info("{}:{}", lister.getPrefix(), record);
+                procedureLogger.info("{}-|-{}", lister.getPrefix(), record);
             }
             if (map != null) map.put("start", lister.currentEndKey());
             if (stopped) break;
@@ -271,7 +271,7 @@ void listing(IStorageLister<E> lister) {
                 processorMap.put(orderStr, lineProcessor);
             }
             export(lister, saver, lineProcessor);
-            procedureLogger.info("{}:", lister.getPrefix());
+            procedureLogger.info("{}-|-", lister.getPrefix());
             progressMap.remove(lister.getPrefix()); // 只有 export 成功情况下才移除 record
         } catch (QiniuException e) {
             try { FileUtils.createIfNotExists(errorLogFile); } catch (IOException ignored) {}

src/main/java/com/qiniu/datasource/FileContainer.java

@@ -253,7 +253,7 @@ public void export(IFileLister<E, File> lister, IResultOutput saver, ILineProces
                 record = json.toString();
                 progressMap.put(lister.getName(), record);
                 try { FileUtils.createIfNotExists(procedureLogFile); } catch (IOException ignored) {}
-                procedureLogger.info("{}:{}", lister.getName(), record);
+                procedureLogger.info("{}-|-{}", lister.getName(), record);
             }
             if (stopped) break;
 //            objects.clear(); 上次其实不能做 clear，会导致 lister 中的列表被清空
@@ -279,7 +279,7 @@ private void listing(IFileLister<E, File> lister) {
                 processorMap.put(orderStr, lineProcessor);
             }
             export(lister, saver, lineProcessor);
-            procedureLogger.info("{}:", lister.getName());
+            procedureLogger.info("{}-|-", lister.getName());
             progressMap.remove(lister.getName()); // 只有 export 成功情况下才移除 record
         }  catch (QiniuException e) {
             try { FileUtils.createIfNotExists(errorLogFile); } catch (IOException ignored) {}

src/main/java/com/qiniu/datasource/TextContainer.java

@@ -166,7 +166,7 @@ public void export(ITextReader<S> reader, IResultOutput saver, ILineProcess<T> p
             record = json.toString();
             progressMap.put(reader.getName(), record);
             try { FileUtils.createIfNotExists(procedureLogFile); } catch (IOException ignored) {}
-            procedureLogger.info("{}:{}", reader.getName(), record);
+            procedureLogger.info("{}-|-{}", reader.getName(), record);
             lastLine = reader.currentEndLine();
         }
     }
@@ -187,7 +187,7 @@ private void reading(ITextReader<S> reader) {
                 processorMap.put(orderStr, lineProcessor);
             }
             export(reader, saver, lineProcessor);
-            procedureLogger.info("{}:", reader.getName());
+            procedureLogger.info("{}-|-", reader.getName());
             progressMap.remove(reader.getName()); // 只有 export 成功情况下才移除 record
         }  catch (QiniuException e) {
             try { FileUtils.createIfNotExists(errorLogFile); } catch (IOException ignored) {}

src/main/java/com/qiniu/entry/CommonParams.java

@@ -44,6 +44,7 @@ public class CommonParams {
     private String baiduAccessId;
     private String baiduSecretKey;
     private String bucket;
+    private String logFilepath;
     private Map<String, Map<String, String>> pathConfigMap;
     private List<String> antiDirectories;
     private List<String> antiPrefixes;
@@ -164,12 +165,12 @@ public CommonParams(IEntryParam entryParam) throws Exception {
         path = entryParam.getValue("path", "");
         setSource();
         accountInit();
+        logFilepath = entryParam.getValue("log", null);
         if (isStorageSource) {
             setAuthKey();
             setBucket();
             String prefixes = entryParam.getValue("prefixes", null);
             setPathConfigMap(entryParam.getValue("prefix-config", ""), prefixes, true, true);
-            antiPrefixes = Arrays.asList(ParamsUtils.escapeSplit(entryParam.getValue("anti-prefixes", "")));
             setPrefixLeft(entryParam.getValue("prefix-left", "false").trim());
             setPrefixRight(entryParam.getValue("prefix-right", "false").trim());
         } else {
@@ -181,6 +182,7 @@ public CommonParams(IEntryParam entryParam) throws Exception {
             String files = entryParam.getValue("files", null);
             setPathConfigMap(entryParam.getValue("file-config", ""), files, false, false);
         }
+        antiPrefixes = Arrays.asList(ParamsUtils.escapeSplit(entryParam.getValue("anti-prefixes", "")));
         setProcess();
         setPrivateType();
         regionName = entryParam.getValue("region", "").trim().toLowerCase();
@@ -297,11 +299,14 @@ public CommonParams(Map<String, String> paramsMap) throws Exception {
                 setSaveSeparator();
                 break;
             case "qupload":
-                if (!fromLine) mapLine.put("key", entryParam.getValue("key", ""));
-                String filepath = entryParam.getValue("filepath", "").trim();
+                String key = entryParam.getValue("key", "");
+                if (!fromLine) mapLine.put("key", key);
+                String filepath = entryParam.getValue("filepath", "");
                 if (!"".equals(filepath)) {
                     indexMap.put("filepath", "filepath");
                     mapLine.put("filepath", filepath);
+                } else if ("".equals(key)) {
+                    throw new IOException("filepath and key shouldn't all be empty, file must be found with them.");
                 }
                 break;
             case "mime":
@@ -602,9 +607,9 @@ private void setProcess() throws Exception {
         }
         if ("qupload".equals(process) && "file".equals(entryParam.getValue("parse", "file")) && !"terminal".equals(source)) {
             isSelfUpload = true;
-            String prefixes = entryParam.getValue("directories", null);
-            setPathConfigMap(entryParam.getValue("directory-config", ""), prefixes, false, true);
-            antiDirectories = Arrays.asList(ParamsUtils.escapeSplit(entryParam.getValue("anti-directories", "")));
+            parse = "file"; // 修正 parse 的默认值
+            String directories = entryParam.getValue("directories", null);
+            setPathConfigMap(entryParam.getValue("directory-config", ""), directories, false, true);
         }
     }
 
@@ -671,63 +676,91 @@ private void setPrivateType() throws IOException {
         }
     }
 
-    private Map<String, String> fromProcedureLog(String logFile) throws IOException {
-        File file = new File(logFile);
-        FileReader fileReader = new FileReader(file);
-        BufferedReader bufferedReader = new BufferedReader(fileReader);
-        Map<String, String> map = new HashMap<>();
-        int index;
-        String line;
-        while ((line = bufferedReader.readLine()) != null) {
-            index = line.indexOf(":{");
-            map.put(line.substring(0, index), line.substring(index));
+    private void fromProcedureLog(String logFile, boolean withMarker, boolean withEnd) throws IOException {
+        String lastLine = FileUtils.lastLineOfFile(logFile);
+        if (lastLine != null && !"".equals(lastLine)) {
+            try {
+                JsonObject jsonObject = JsonUtils.toJsonObject(lastLine);
+                parseConfigMapFromJson(jsonObject, withMarker, withEnd);
+            } catch (Exception e) {
+                File file = new File(logFile);
+                FileReader fileReader = new FileReader(file);
+                BufferedReader bufferedReader = new BufferedReader(fileReader);
+                int index;
+                String line;
+                String value;
+                Map<String, String> map = new HashMap<>();
+                while ((line = bufferedReader.readLine()) != null) {
+                    index = line.indexOf("-|-");
+                    map.put(line.substring(0, index), line.substring(index));
+                }
+                for (String key : map.keySet()) {
+                    value = map.get(key);
+                    if (!"".equals(value)) {
+                        pathConfigMap.put(key, JsonUtils.fromJson(value, map.getClass()));
+                    }
+                }
+            }
         }
-        return map;
     }
 
-    private void setPathConfigMap(String jsonConfigPath, String subPaths, boolean withMarker, boolean withEnd) throws Exception {
-        pathConfigMap = new HashMap<>();
-        if (jsonConfigPath != null && !"".equals(jsonConfigPath)) {
-            JsonFile jsonFile = new JsonFile(jsonConfigPath);
-            JsonObject jsonCfg;
-            JsonElement markerElement;
-            JsonElement startElement;
-            JsonElement endElement;
-            for (String key : jsonFile.getKeys()) {
-                Map<String, String> startAndEnd = new HashMap<>();
+    private void parseConfigMapFromJson(JsonObject jsonObject, boolean withMarker, boolean withEnd) throws IOException {
+        JsonObject jsonCfg;
+        JsonElement markerElement;
+        JsonElement startElement;
+        JsonElement endElement;
+        for (String key : jsonObject.keySet()) {
+            Map<String, String> startAndEnd = new HashMap<>();
 //                if ("".equals(prefix)) throw new IOException("prefix (prefixes config's element key) can't be empty.");
-                JsonElement json = jsonFile.getElement(key);
-                if (json == null || json instanceof JsonNull) {
-                    pathConfigMap.put(key, null);
-                    continue;
-                }
-                if (withMarker || withEnd) {
-                    if (!(json instanceof JsonObject)) throw new IOException("the value of key: " + key + " must be json.");
-                    jsonCfg = json.getAsJsonObject();
-                    if (withMarker) {
-                        markerElement = jsonCfg.get("marker");
-                        if (markerElement != null && !(markerElement instanceof JsonNull)) {
-                            startAndEnd.put("marker", markerElement.getAsString());
-                        }
-                    }
-                    startElement = jsonCfg.get("start");
-                    if (startElement != null && !(startElement instanceof JsonNull)) {
-                        startAndEnd.put("start", startElement.getAsString());
+            JsonElement json = jsonObject.get(key);
+            if (json == null || json instanceof JsonNull) {
+                pathConfigMap.put(key, null);
+                continue;
+            }
+            if (withMarker || withEnd) {
+                if (!(json instanceof JsonObject)) throw new IOException("the value of key: " + key + " must be json.");
+                jsonCfg = json.getAsJsonObject();
+                if (withMarker) {
+                    markerElement = jsonCfg.get("marker");
+                    if (markerElement != null && !(markerElement instanceof JsonNull)) {
+                        startAndEnd.put("marker", markerElement.getAsString());
                     }
-                    if (withEnd) {
-                        endElement = jsonCfg.get("end");
-                        if (endElement != null && !(endElement instanceof JsonNull)) {
-                            startAndEnd.put("end", endElement.getAsString());
-                        }
+                }
+                startElement = jsonCfg.get("start");
+                if (startElement != null && !(startElement instanceof JsonNull)) {
+                    startAndEnd.put("start", startElement.getAsString());
+                }
+                if (withEnd) {
+                    endElement = jsonCfg.get("end");
+                    if (endElement != null && !(endElement instanceof JsonNull)) {
+                        startAndEnd.put("end", endElement.getAsString());
                     }
-                } else {
-                    startAndEnd.put("start", json.getAsString());
                 }
-                pathConfigMap.put(key, startAndEnd);
+            } else {
+                startAndEnd.put("start", json.getAsString());
+            }
+            pathConfigMap.put(key, startAndEnd);
+        }
+    }
+
+    private void setPathConfigMap(String jsonConfigPath, String subPaths, boolean withMarker, boolean withEnd) throws Exception {
+        pathConfigMap = new HashMap<>();
+        if (logFilepath == null || "".equals(logFilepath)) {
+            if (jsonConfigPath != null && !"".equals(jsonConfigPath)) {
+                JsonFile jsonFile = new JsonFile(jsonConfigPath);
+                parseConfigMapFromJson(jsonFile.getJsonObject(), withMarker, withEnd);
+            } else if (subPaths != null && !"".equals(subPaths)) {
+                String[] subPathList = ParamsUtils.escapeSplit(subPaths);
+                for (String subPath : subPathList) pathConfigMap.put(subPath, null);
+            }
+        } else {
+            if (jsonConfigPath != null && !"".equals(jsonConfigPath)) {
+                throw new IOException("log and uris can not be used together, please remove prefixes/files/directories if you want use breakpoint with log.");
+            } else if (subPaths != null && !"".equals(subPaths)) {
+                throw new IOException("log and json config can not be used together, please remove config path if you want use breakpoint with log.");
+            } else {
+                fromProcedureLog(logFilepath, withMarker, withEnd);
             }
-        } else if (subPaths != null && !"".equals(subPaths)) {
-            String[] subPathList = ParamsUtils.escapeSplit(subPaths);
-            for (String subPath : subPathList) pathConfigMap.put(subPath, null);
         }
     }
 
@@ -991,8 +1024,8 @@ private void setIndexMap() throws IOException {
                 if (!indexMap.containsKey("0")) indexMap.put("0", "key");
             }
         }
-        if (ProcessUtils.needFilepath(process) || "file".equals(parse) || isSelfUpload) {
-            setIndex(entryParam.getValue("filepath-index", "filepath").trim(), "filepath");
+        if (ProcessUtils.needFilepath(process) || "file".equals(parse)) {
+            setIndex(entryParam.getValue("filepath-index", fieldIndex ? "filepath" : "").trim(), "filepath");
 //            setIndex("parent", "parent");
         }
         if (indexMap.size() == 0) {
@@ -1097,7 +1130,7 @@ private void checkFilterForProcess() throws IOException {
 
     private void setUnitLen(String unitLen) throws IOException {
         if (unitLen.startsWith("-")) {
-            if (isSelfUpload) this.unitLen = 20;
+            if (isSelfUpload) this.unitLen = 3;
             if ("qiniu".equals(source) || "local".equals(source)) this.unitLen = 10000;
             else this.unitLen = 1000;
         } else {

src/main/java/com/qiniu/entry/QSuitsEntry.java

@@ -949,8 +949,7 @@ private ILineProcess<Map<String, String>> getCensorResult(Map<String, String> in
                 : new QueryCensorResult(qiniuAccessKey, qiniuSecretKey, getQiniuConfig(), jobIdIndex, savePath);
     }
 
-    private ILineProcess<Map<String, String>> getQiniuUploadFile(Map<String, String> indexMap, boolean single)
-            throws IOException {
+    private ILineProcess<Map<String, String>> getQiniuUploadFile(Map<String, String> indexMap, boolean single) throws IOException {
         String pathIndex = indexMap.containsValue("filepath") ? "filepath" : null;
         String parentPath = entryParam.getValue("parent-path", "").trim();
         String recorder = entryParam.getValue("record", "false").trim();

src/main/java/com/qiniu/util/FileUtils.java

@@ -1,11 +1,10 @@
 package com.qiniu.util;
 
 import javax.activation.MimetypesFileTypeMap;
-import java.io.File;
-import java.io.IOException;
-import java.io.RandomAccessFile;
+import java.io.*;
 import java.net.FileNameMap;
 import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
@@ -225,4 +224,27 @@ public static void randomModify(String filePath, String oldStr, String newStr) t
             }
         }
     }
+
+    public static String lastLineOfFile(String filePath) throws IOException {
+        RandomAccessFile accessFile = new RandomAccessFile(filePath, "r");
+        if (accessFile.length() < 1) return null;
+        byte[] bytes = new byte[1];
+        long pos = accessFile.length();
+        if (pos > 2) {
+            accessFile.seek(--pos);
+            accessFile.read(bytes);
+            String s = new String(bytes);
+            accessFile.seek(--pos);
+            while (pos > 0 && accessFile.read(bytes) != -1 && "\n".equals(s)) {
+                accessFile.seek(--pos);
+                s = new String(bytes);
+            }
+            while (pos > 0 && accessFile.read(bytes) != -1 && !"\n".equals(s)) {
+                accessFile.seek(--pos);
+                s = new String(bytes);
+            }
+            if (pos > 0) accessFile.seek(pos + 2);
+        }
+        return new String(accessFile.readLine().getBytes(StandardCharsets.ISO_8859_1));
+    }
 }