git.useribm.hu - mediacube.git/blob

   1 package user.jobengine.server.steps;\r
   2 \r
   3 import java.io.File;\r
   4 import java.io.IOException;\r
   5 import java.nio.charset.Charset;\r
   6 import java.nio.file.DirectoryStream;\r
   7 import java.nio.file.Files;\r
   8 import java.nio.file.Path;\r
   9 import java.nio.file.Paths;\r
  10 import java.text.SimpleDateFormat;\r
  11 import java.util.ArrayList;\r
  12 import java.util.Arrays;\r
  13 import java.util.Date;\r
  14 import java.util.HashMap;\r
  15 import java.util.List;\r
  16 import java.util.Locale;\r
  17 import java.util.Map;\r
  18 import java.util.Set;\r
  19 \r
  20 import org.apache.logging.log4j.LogManager;\r
  21 import org.apache.logging.log4j.Logger;\r
  22 \r
  23 import com.ibm.nosql.json.api.BasicDBObject;\r
  24 import com.ibm.nosql.json.api.DB;\r
  25 import com.ibm.nosql.json.api.DBCollection;\r
  26 import com.ibm.nosql.json.api.DBObject;\r
  27 import com.ibm.nosql.json.api.QueryBuilder;\r
  28 \r
  29 import user.commons.IEntityBase;\r
  30 import user.commons.nosql.NoSQLUtils;\r
  31 import user.jobengine.db.IItemManager;\r
  32 import user.jobengine.db.MediaFile;\r
  33 import user.jobengine.db.MediaFileDAO;\r
  34 import user.jobengine.server.IJobEngine;\r
  35 import user.jobengine.server.IJobRuntime;\r
  36 \r
  37 public class ImportMORPHEUSMissingMaterialsStep extends JobStep {\r
  38         private static final String STATUS = "Status";\r
  39         private static final String IMPORTED = "Imported";\r
  40         private static final Logger logger = LogManager.getLogger();\r
  41         private static final String MATERIAL_ID = "Material ID";\r
  42         private static final String CHANNEL = "Channel";\r
  43         private static final String TIME_TO_AIR = "Time to Air";\r
  44         private static final String DURATION = "Duration";\r
  45         private static final String TITLE = "Title";\r
  46         private static final String DEVICE_ID = "Device ID";\r
  47         private static final String REASON = "Reason";\r
  48         private static final String COLLECTION_NAME = "missing_materials";\r
  49 \r
  50         private static final String CSV_EXT = ".csv";\r
  51         private static final String MXF_EXT = ".MXF";\r
  52         private MediaFileDAO dao;\r
  53         private String processedFolder;\r
  54         private DB db;\r
  55         private String targetPath;\r
  56         private IJobRuntime jobRuntime;\r
  57         private int overall;\r
  58         private int current;\r
  59         private final SimpleDateFormat enDateFormat = new SimpleDateFormat("dd-MMM-yyyy HH:mm:ss:S", Locale.ENGLISH);\r
  60 \r
  61         private Map<String, Integer> buildMetadataMap(Path csvFilePath, String[] data) throws Exception {\r
  62                 Map<String, Integer> result = new HashMap<>();\r
  63                 List<String> dataList = Arrays.asList(data);\r
  64                 storeMetadataPosition(csvFilePath, dataList, MATERIAL_ID, result);\r
  65                 storeMetadataPosition(csvFilePath, dataList, CHANNEL, result);\r
  66                 storeMetadataPosition(csvFilePath, dataList, TIME_TO_AIR, result);\r
  67                 storeMetadataPosition(csvFilePath, dataList, DURATION, result);\r
  68                 storeMetadataPosition(csvFilePath, dataList, TITLE, result);\r
  69                 storeMetadataPosition(csvFilePath, dataList, DEVICE_ID, result);\r
  70                 storeMetadataPosition(csvFilePath, dataList, REASON, result);\r
  71                 return result;\r
  72         }\r
  73 \r
  74         @StepEntry\r
  75         public Object[] execute(String sourceCSVPath, String processedFolder, String targetPath, IJobEngine jobEngine, IJobRuntime jobRuntime) throws Exception {\r
  76                 this.jobRuntime = jobRuntime;\r
  77                 setAndCheck(sourceCSVPath, processedFolder, targetPath, jobEngine);\r
  78 \r
  79                 List<Path> filePaths = new ArrayList<>();\r
  80                 try (DirectoryStream<Path> directoryStream = Files.newDirectoryStream(Paths.get(sourceCSVPath))) {\r
  81                         for (Path path : directoryStream) {\r
  82                                 filePaths.add(path);\r
  83                         }\r
  84 \r
  85                         processPathItems(filePaths);\r
  86                 } catch (Exception e) {\r
  87                         logger.catching(e);\r
  88                         logger.error(jobRuntime.getMarker(), "Hiba a végrehajtás során. A rendszer üzenete: {}", e.getMessage());\r
  89                 }\r
  90                 return null;\r
  91         }\r
  92 \r
  93         private void moveProcessedCSV(Path csvFilePath) throws IOException {\r
  94                 EscortFiles.ensureUNCFolder(csvFilePath.getParent().toString(), processedFolder);\r
  95                 String fileName = csvFilePath.getFileName() + "." + EscortFiles.composeKillDate(0);\r
  96                 Path targetPath = Paths.get(csvFilePath.getParent().toString(), processedFolder, fileName);\r
  97                 Files.move(csvFilePath, targetPath);\r
  98         }\r
  99 \r
 100         private void processLine(String[] data, Map<String, Integer> metadatas) throws Exception {\r
 101                 String channel = data[metadatas.get(CHANNEL)];\r
 102                 String timeToAir = data[metadatas.get(TIME_TO_AIR)];\r
 103                 String duration = data[metadatas.get(DURATION)];\r
 104                 String materialID = data[metadatas.get(MATERIAL_ID)];\r
 105                 String title = data[metadatas.get(TITLE)];\r
 106                 String deviceID = data[metadatas.get(DEVICE_ID)];\r
 107                 String reason = data[metadatas.get(REASON)];\r
 108 \r
 109                 DBObject query = QueryBuilder.start().and(QueryBuilder.start(MATERIAL_ID).is(materialID).get(), QueryBuilder.start(TIME_TO_AIR).is(timeToAir).get())\r
 110                                 .get();\r
 111                 DBCollection collection = db.getCollection(COLLECTION_NAME);\r
 112                 BasicDBObject existingObject = (BasicDBObject) collection.findOne(query);\r
 113                 if (existingObject != null) {\r
 114                         logger.warn(jobRuntime.getMarker(), "Az '{}' anyag már feldolgozásra került az {} időpontban.", materialID, existingObject.getDate(IMPORTED));\r
 115                         return;\r
 116                 }\r
 117 \r
 118                 BasicDBObject dbObject = new BasicDBObject(IMPORTED, new Date());\r
 119                 dbObject.put(CHANNEL, channel);\r
 120                 dbObject.put(TIME_TO_AIR.replace(" ", ""), enDateFormat.parse(timeToAir));\r
 121                 dbObject.put(DURATION, duration);\r
 122                 dbObject.put(MATERIAL_ID.replace(" ", ""), materialID);\r
 123                 dbObject.put(TITLE, title);\r
 124                 dbObject.put(DEVICE_ID.replace(" ", ""), deviceID);\r
 125                 dbObject.put(REASON, reason);\r
 126 \r
 127                 String fileName = materialID + MXF_EXT;\r
 128                 if (Files.exists(Paths.get(targetPath, fileName))) {\r
 129                         logger.warn(jobRuntime.getMarker(), "Az '{}' anyag már be van töltve.", materialID);\r
 130                         dbObject.put(STATUS, "SKIPPED");\r
 131                 } else {\r
 132                         List<IEntityBase> medias = dao.getByHouseId(fileName);\r
 133                         if (medias == null || medias.size() == 0) {\r
 134                                 logger.warn(jobRuntime.getMarker(), "Az '{}' anyag nem található az archívumban.", materialID);\r
 135                                 dbObject.put(STATUS, "UNAVAILABLE");\r
 136                         } else if (medias.size() > 1) {\r
 137                                 logger.warn(jobRuntime.getMarker(), "Az '{}' anyagból egynél több található az archívumban.", materialID);\r
 138                                 dbObject.put(STATUS, "MULTIPLE");\r
 139                         } else {\r
 140                                 logger.info(jobRuntime.getMarker(), "Az '{}' anyag megtalálható az archívumban.", materialID);\r
 141                                 dbObject.put(STATUS, "RESTORABLE");\r
 142                         }\r
 143 \r
 144                 }\r
 145 \r
 146                 collection.insert(dbObject);\r
 147         }\r
 148 \r
 149         //      Channel,Time to Air,Duration,Material ID,Title,Device ID,Reason,\r
 150         //      TX02,10-JAN-2018 13:25:21:08,00:05:00:00,M009572A,Tiéd a pálya/26. - 1. seg - Eredeti ** mc ,ICELE-02,On Domain (ISILON, ICELE-01, ICELE-05) ,\r
 151         private void processMissingMaterialCSV(Path csvFilePath, List<String> lines) throws Exception {\r
 152                 if (lines == null | lines.size() == 0) {\r
 153                         return;\r
 154                 }\r
 155 \r
 156                 Map<String, Integer> metadatas = null;\r
 157                 for (int i = 0; i < lines.size(); i++) {\r
 158                         String line = lines.get(i);\r
 159                         if (line == null)\r
 160                                 continue;\r
 161                         String[] data = line.split(",");\r
 162                         if (i == 0)\r
 163                                 metadatas = buildMetadataMap(csvFilePath, data);\r
 164                         else\r
 165                                 processLine(data, metadatas);\r
 166 \r
 167                         current++;\r
 168                         setProgress(current * 100 / overall);\r
 169                 }\r
 170 \r
 171         }\r
 172 \r
 173         private void processPathItem(Path csvFilePath, List<String> lines) {\r
 174                 File csvFile = csvFilePath.toFile();\r
 175 \r
 176                 try {\r
 177                         processMissingMaterialCSV(csvFilePath, lines);\r
 178                         moveProcessedCSV(csvFilePath);\r
 179                 } catch (Exception e) {\r
 180                         logger.catching(e);\r
 181                         logger.error(jobRuntime.getMarker(), "A {} MORPHEUS állomány mozgatásakor hiba történt. A rendszer hibaüzenete: {}.", csvFile.getName(),\r
 182                                         e.getMessage());\r
 183                 }\r
 184         }\r
 185 \r
 186         private void processPathItems(List<Path> filePaths) throws IOException {\r
 187                 overall = 0;\r
 188                 Map<Path, List<String>> contents = new HashMap<>();\r
 189                 for (Path filePath : filePaths) {\r
 190                         File csvFile = filePath.toFile();\r
 191                         if (csvFile.isDirectory() || !csvFile.getName().toLowerCase().endsWith(CSV_EXT.toLowerCase()))\r
 192                                 continue;\r
 193                         List<String> lines = Files.readAllLines(filePath, Charset.forName("UTF-8"));\r
 194                         overall += lines.size();\r
 195                         contents.put(filePath, lines);\r
 196                 }\r
 197 \r
 198                 Set<Path> csvPaths = contents.keySet();\r
 199                 for (Path csvPath : csvPaths) {\r
 200                         processPathItem(csvPath, contents.get(csvPath));\r
 201                 }\r
 202         }\r
 203 \r
 204         private void setAndCheck(String sourcePath, String processedFolder, String targetPath, IJobEngine jobEngine) {\r
 205                 if (jobEngine == null) {\r
 206                         logger.error(jobRuntime.getMarker(), "Az folyamatkezelő réteg nem elérhető.");\r
 207                         throw new NullPointerException("Internal error, missing JobEngine reference.");\r
 208                 }\r
 209 \r
 210                 IItemManager manager = jobEngine.getItemManager();\r
 211                 if (manager == null) {\r
 212                         logger.error(jobRuntime.getMarker(), "Az adatbáziskezelő réteg nem elérhető.");\r
 213                         throw new NullPointerException("Internal error, missing ItemManager reference.");\r
 214                 }\r
 215                 dao = (MediaFileDAO) manager.getBaseDAO(MediaFile.class);\r
 216                 if (dao == null) {\r
 217                         logger.error(jobRuntime.getMarker(), "Az adatbáziskezelő réteg MediaFile kezelöje nem elérhető.");\r
 218                         throw new NullPointerException("Internal error, missing MediaFile DAO reference.");\r
 219                 }\r
 220                 if (sourcePath == null) {\r
 221                         logger.error(jobRuntime.getMarker(), "A folyamat 'sourcePath' bemeneti paramétere üres.");\r
 222                         throw new NullPointerException("System is not configured properly, 'sourcePath' input parameter missing.");\r
 223                 }\r
 224                 if (processedFolder == null) {\r
 225                         logger.error(jobRuntime.getMarker(), "A folyamat 'processedFolder' bemeneti paramétere üres.");\r
 226                         throw new NullPointerException("System is not configured properly, 'processedFolder' input parameter missing.");\r
 227                 }\r
 228                 this.processedFolder = processedFolder;\r
 229 \r
 230                 if (targetPath == null) {\r
 231                         logger.error(jobRuntime.getMarker(), "A folyamat 'targetPath' bemeneti paramétere üres.");\r
 232                         throw new NullPointerException("System is not configured properly, 'targetPath' input parameter missing.");\r
 233                 }\r
 234                 this.targetPath = targetPath;\r
 235 \r
 236                 db = NoSQLUtils.getNoSQLDB();\r
 237                 if (db == null) {\r
 238                         logger.error(jobRuntime.getMarker(), "Sikertelen kapcsolódás a NoSQL adatbázishoz.");\r
 239                         throw new NullPointerException("Can not connect to NoSQL database.");\r
 240                 }\r
 241 \r
 242         }\r
 243 \r
 244         private void storeMetadataPosition(Path csvFilePath, List<String> dataList, String name, Map<String, Integer> metadatas) throws Exception {\r
 245                 int pos = dataList.indexOf(name);\r
 246                 if (pos < 0)\r
 247                         throw new Exception(String.format("A '%s' MORPHEUS állományban nem található a '%s' mező.", csvFilePath.getFileName(), MATERIAL_ID));\r
 248                 metadatas.put(name, pos);\r
 249         }\r
 250 \r
 251 }\r