git.useribm.hu - mediacube.git/blob

   1 package user.jobengine.server.steps;\r
   2 \r
   3 import java.io.OutputStream;\r
   4 import java.net.InetAddress;\r
   5 import java.net.URI;\r
   6 import java.util.Calendar;\r
   7 import java.util.Date;\r
   8 import java.util.List;\r
   9 \r
  10 import org.apache.commons.lang.StringUtils;\r
  11 import org.apache.commons.net.ftp.FTP;\r
  12 import org.apache.commons.net.ftp.FTPClient;\r
  13 import org.apache.commons.net.ftp.FTPReply;\r
  14 import org.apache.logging.log4j.LogManager;\r
  15 import org.apache.logging.log4j.Logger;\r
  16 \r
  17 import com.ibm.nosql.json.api.BasicDBObject;\r
  18 import com.ibm.nosql.json.api.DB;\r
  19 import com.ibm.nosql.json.api.DBCollection;\r
  20 import com.ibm.nosql.json.api.DBCursor;\r
  21 import com.ibm.nosql.json.api.DBObject;\r
  22 \r
  23 import user.commons.CalendarUtils;\r
  24 import user.commons.ListUtils;\r
  25 import user.commons.StoreUri;\r
  26 import user.commons.nosql.NoSQLUtils;\r
  27 import user.commons.octopus.IOctopusAPI;\r
  28 import user.commons.octopus.OctopusAPI;\r
  29 import user.commons.remotestore.FtpDirectoryLister;\r
  30 import user.commons.remotestore.RemoteStoreProtocol;\r
  31 import user.jobengine.db.IItemManager;\r
  32 import user.jobengine.server.IJobEngine;\r
  33 import user.jobengine.server.IJobRuntime;\r
  34 import user.jobengine.server.steps.MetadataTypeDetector.MetadataType;\r
  35 \r
  36 public class CopyForArchiveNEXIOMaterialsStep extends JobStep {\r
  37         private static final String SCHEDULED_FORMAT = "yyyy.MM.dd HH:mm";\r
  38         private static final Logger logger = LogManager.getLogger();\r
  39         private static final String UTF_8 = "utf-8";\r
  40         private static final String JSON_EXT = ".json";\r
  41         private static final String XML_EXT = ".xml";\r
  42         private static final String DURATION = "duration";\r
  43         private static final String MXFEXT = ".MXF";\r
  44         private static final String NEXIOCLIPS = "nexioclips";\r
  45         private static final String LONGNAMEID = "longnameid";\r
  46         private static final String ARCHIVEDRUNDOWNS = "archivedrundowns";\r
  47         private static final String ID = "id";\r
  48         private OctopusAPI octopusAPI;\r
  49         private IItemManager manager;\r
  50 \r
  51         private DB db;\r
  52         private FTPClient sourceFtp;\r
  53         private FTPClient targetFtp;\r
  54         private StoreUri sourceUri;\r
  55         private StoreUri targetUri;\r
  56         private int nexioKillDateDays;\r
  57         private String nexioAgency;\r
  58 \r
  59         private int check(int value, String name) {\r
  60                 if (value == 0) {\r
  61                         logger.error(getMarker(), "A folyamat '{}' bemeneti paramétere 0.", name);\r
  62                         throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));\r
  63                 }\r
  64                 return value;\r
  65         }\r
  66 \r
  67         private String check(String value, String name) {\r
  68                 if (value == null) {\r
  69                         logger.error(getMarker(), "A folyamat '{}' bemeneti paramétere üres.", name);\r
  70                         throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));\r
  71                 }\r
  72                 return value;\r
  73         }\r
  74 \r
  75         private void copy(RundownArchive rundownArchive) throws Exception {\r
  76                 for (StoryArchive storyArchive : rundownArchive.getStoryArchives()) {\r
  77                         for (FileArchive fileArchive : storyArchive.getFileArchives()) {\r
  78                                 copyFile(fileArchive, rundownArchive, storyArchive);\r
  79                         }\r
  80                 }\r
  81         }\r
  82 \r
  83         private void copyFile(FileArchive fileArchive, RundownArchive rundownArchive, StoryArchive storyArchive) throws Exception {\r
  84                 String fileName = fileArchive.getFileName();\r
  85                 //TODO file existance\r
  86                 //              targetFtp = ((FtpDirectoryLister) RemoteFileHandler.createLister(targetUri)).connect();\r
  87                 //              int version = 1;\r
  88                 //              while (true) {\r
  89                 //                      targetFtp.listNames(arg0)\r
  90                 //              }\r
  91                 //\r
  92                 //              if (targetUri != null)\r
  93                 //                      targetUri.cleanUp();\r
  94                 String videoFileName = fileName + MXFEXT;\r
  95 \r
  96                 transferFile(videoFileName);\r
  97                 BasicDBObject metadata = createMetadata(rundownArchive, storyArchive, fileArchive);\r
  98                 transferMetadata(videoFileName, metadata);\r
  99                 //targetFtp.changeToParentDirectory();\r
 100                 createSourceKillDateFile(rundownArchive, fileName);\r
 101 \r
 102         }\r
 103 \r
 104         private BasicDBObject createMetadata(RundownArchive rundownArchive, StoryArchive storyArchive, FileArchive fileArchive) {\r
 105                 //              {\r
 106                 //                        "itemHouseId": "43",\r
 107                 //                        "itemTitle": "Hazahúzó",\r
 108                 //                        "itemDescription": null,\r
 109                 //                        "mediaHouseId": "002570",\r
 110                 //                        "mediaTitle": "2017.12.13",\r
 111                 //                        "mediaDescription": null,\r
 112                 //                        "ok": false,\r
 113                 //                        "userName": "echotest"\r
 114                 //              }\r
 115                 BasicDBObject result = new BasicDBObject();\r
 116                 result.put("itemHouseID", rundownArchive.getItemHouseId());\r
 117                 result.put("itemTitle", rundownArchive.getItemTitle());\r
 118                 result.put("itemDescription", rundownArchive.getItemDesc());\r
 119                 result.put("userName", "mediacube");\r
 120 \r
 121                 result.put("mediaHouseId", storyArchive.getMediaHouseId());\r
 122                 result.put("mediaTitle", storyArchive.getMediaTitle());\r
 123                 result.put("mediaDescription", storyArchive.getMediaDesc());\r
 124 \r
 125                 result.put("duration", fileArchive.getDuration());\r
 126                 return result;\r
 127         }\r
 128 \r
 129         private void createSourceKillDateFile(RundownArchive rundownArchive, String fileName) throws Exception {\r
 130                 //sourceFtp = ((FtpDirectoryLister) RemoteFileHandler.createLister(sourceUri)).connect();\r
 131                 OutputStream outStream = null;\r
 132                 try {\r
 133                         sourceFtp.enterLocalActiveMode();\r
 134                         Calendar killDate = CalendarUtils.createCalendar(rundownArchive.getScheduleDate());\r
 135                         killDate.add(Calendar.DAY_OF_YEAR, nexioKillDateDays);\r
 136                         byte[] killDateFile = EscortFiles.createNEXIOKillDateFile(fileName, killDate.getTime(), null, nexioAgency);\r
 137                         outStream = sourceFtp.storeFileStream(fileName + XML_EXT);\r
 138                         if (outStream == null) {\r
 139                                 throw new NullPointerException("Can not open: " + fileName + XML_EXT + " Reply:" + sourceFtp.getReplyString());\r
 140                         }\r
 141                         outStream.write(killDateFile);\r
 142                         outStream.flush();\r
 143                 } catch (Exception e) {\r
 144                         logger.catching(e);\r
 145                         throw e;\r
 146                 } finally {\r
 147                         if (outStream != null)\r
 148                                 outStream.close();\r
 149                         //                      if (sourceUri != null)\r
 150                         //                              sourceUri.cleanUp();\r
 151                 }\r
 152         }\r
 153 \r
 154         @StepEntry\r
 155         public Object[] execute(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,\r
 156                         int daysBeforeNow, int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine, IJobRuntime jobRuntime) throws Exception {\r
 157                 setAndCheck(nexioPort, nexioUserName, nexioPassword, archiveFtp, archiveUserName, archivePassword, nexioKillDateDays, nexioAgency, jobEngine);\r
 158                 octopusAPI = new OctopusAPI();\r
 159                 Calendar scheduledDate = Calendar.getInstance();\r
 160                 scheduledDate.add(Calendar.DAY_OF_YEAR, -1 * daysBeforeNow);\r
 161                 List<DBObject> rundowns = octopusAPI.getRundowns(scheduledDate.getTime());\r
 162                 if (rundowns == null) {\r
 163                         logger.warn(getMarker(), "Nem található adástükör a {} napra.", CalendarUtils.toDateString(scheduledDate));\r
 164                         return null;\r
 165                 }\r
 166 \r
 167                 sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();\r
 168                 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
 169                 processRundowns(rundowns);\r
 170                 if (sourceUri != null)\r
 171                         sourceUri.cleanUp();\r
 172                 if (targetUri != null)\r
 173                         targetUri.cleanUp();\r
 174                 return null;\r
 175         }\r
 176 \r
 177         private FileArchive processMosObject(BasicDBObject rundown, BasicDBObject story, BasicDBObject mosObject) throws Exception {\r
 178                 String mosID = mosObject.getString(IOctopusAPI.OBJ_ID);\r
 179                 if (MetadataTypeDetector.GuessMetadataType(mosID) != MetadataType.OCTOPUSPLACEHOLDER) {\r
 180                         logger.trace("Skipping MOS object {}", mosID);\r
 181                         return null;\r
 182                 }\r
 183                 DBCollection clips = db.getCollection(NEXIOCLIPS);\r
 184                 BasicDBObject clip = (BasicDBObject) clips.findOne(new BasicDBObject(LONGNAMEID, mosID));\r
 185                 if (clip == null) {\r
 186                         logger.debug("File NOT exists {}", mosID);\r
 187                         throw new Exception(String.format("File NOT exists %s", mosID));\r
 188                 } else {\r
 189                         logger.debug("File exists {}", mosID);\r
 190                 }\r
 191                 long duration = NoSQLUtils.asLong(clip, DURATION);\r
 192                 return new FileArchive(mosID, duration);\r
 193         }\r
 194 \r
 195         private RundownArchive processRundow(DBObject r) throws Exception {\r
 196                 BasicDBObject rundown = (BasicDBObject) r;\r
 197                 long rundownID = rundown.getLong(ID);\r
 198                 logger.info("Processing rundown {} {}", rundownID, rundown.getString(IOctopusAPI.NAME));\r
 199                 List<DBObject> stories = octopusAPI.getRundownFullStories(rundownID);\r
 200                 if (stories == null)\r
 201                         return null;\r
 202                 RundownArchive result = new RundownArchive();\r
 203 \r
 204                 long id = NoSQLUtils.asLong(rundown, IOctopusAPI.ID);\r
 205                 if (id == 0)\r
 206                         return null;\r
 207                 String name = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.RUNDOWN_TYPE), IOctopusAPI.NAME);\r
 208                 if (StringUtils.isBlank(name))\r
 209                         return null;\r
 210                 String channel = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.CHANNEL), IOctopusAPI.NAME);\r
 211                 Date scheduledStart = rundown.getDate(IOctopusAPI.SCHEDULED_START);\r
 212                 if (scheduledStart == null)\r
 213                         return null;\r
 214                 String start = CalendarUtils.toString(CalendarUtils.createCalendar(scheduledStart), SCHEDULED_FORMAT);\r
 215                 result.setScheduleDate(scheduledStart);\r
 216                 result.setItemHouseId(String.valueOf(id));\r
 217                 result.setItemTitle(String.format("%s %s %s", start, name, channel));\r
 218 \r
 219                 for (DBObject s : stories) {\r
 220                         StoryArchive storyArchive = processStory(rundown, s);\r
 221                         if (storyArchive == null)\r
 222                                 continue;\r
 223                         result.addStoryArchive(storyArchive);\r
 224                 }\r
 225                 return result;\r
 226         }\r
 227 \r
 228         private void processRundowns(List<DBObject> rundowns) {\r
 229                 db.getCollection(ARCHIVEDRUNDOWNS).drop();\r
 230                 List<BasicDBObject> archivedRundowns = queryArchivedRundowns();\r
 231 \r
 232                 int index = 1;\r
 233 \r
 234                 for (DBObject r : rundowns) {\r
 235                         BasicDBObject rundown = (BasicDBObject) r;\r
 236                         setProgress(index * 100 / rundowns.size());\r
 237                         try {\r
 238                                 long rundownID = NoSQLUtils.asLong(rundown, IOctopusAPI.ID);\r
 239                                 BasicDBObject currentRundownID = new BasicDBObject(IOctopusAPI.ID, rundownID);\r
 240                                 String rundownName = rundown.getString(IOctopusAPI.NAME);\r
 241                                 if (archivedRundowns != null && archivedRundowns.contains(currentRundownID)) {\r
 242                                         logger.info("Skipping archived rundown {} {}", rundownID, rundownName);\r
 243                                         continue;\r
 244                                 }\r
 245 \r
 246                                 RundownArchive rundownArchive = processRundow(r);\r
 247                                 if (rundownArchive == null || rundownArchive.isEmpty()) {\r
 248                                         logger.info("Skipping rundown {} {}", NoSQLUtils.asLong(rundown, IOctopusAPI.ID), rundown.getString(IOctopusAPI.NAME));\r
 249                                         continue;\r
 250                                 }\r
 251 \r
 252                                 logger.info("Saving rundown {} {}", rundownID, rundownName);\r
 253                                 copy(rundownArchive);\r
 254 \r
 255                                 db.getCollection(ARCHIVEDRUNDOWNS).save(currentRundownID);\r
 256                         } catch (Exception e) {\r
 257                                 logger.catching(e);\r
 258                                 logger.error(getMarker(),\r
 259                                                 "A {} tükör archiválása nem lehetséges, mert a annak ellenőrzése hibát jelzett. A rendszer üzenete: " + e.getMessage());\r
 260                         }\r
 261                         index++;\r
 262                 }\r
 263 \r
 264         }\r
 265 \r
 266         private StoryArchive processStory(BasicDBObject rundown, DBObject s) throws Exception {\r
 267                 BasicDBObject story = (BasicDBObject) s;\r
 268                 String parentStoryID = story.getString(IOctopusAPI.PARENT_STORY_ID);\r
 269                 if (StringUtils.isBlank(parentStoryID)) {\r
 270                         logger.warn("Story parentStoryID is null: {}", story.toPrettyString(null));\r
 271                         return null;\r
 272                 } else\r
 273                         logger.debug("Processing story {}", parentStoryID);\r
 274                 List<BasicDBObject> mosObjects = NoSQLUtils.asList(story, IOctopusAPI.MOS_OBJECTS);\r
 275                 if (mosObjects == null)\r
 276                         return null;\r
 277                 StoryArchive storyArchive = null;\r
 278                 for (BasicDBObject mosObject : mosObjects) {\r
 279                         FileArchive fileArchive = processMosObject(rundown, story, mosObject);\r
 280                         if (fileArchive == null)\r
 281                                 continue;\r
 282                         if (storyArchive == null) {\r
 283                                 storyArchive = new StoryArchive();\r
 284                                 storyArchive.setMediaHouseId(parentStoryID);\r
 285                                 storyArchive.setMediaTitle(story.getString(IOctopusAPI.NAME));\r
 286                                 storyArchive.setMediaDesc(story.getString(IOctopusAPI.SCRIPT_CONTENT));\r
 287 \r
 288                         }\r
 289                         storyArchive.addFileArchive(fileArchive);\r
 290                 }\r
 291                 return storyArchive;\r
 292 \r
 293         }\r
 294 \r
 295         private List<BasicDBObject> queryArchivedRundowns() {\r
 296                 List<BasicDBObject> result = null;\r
 297                 DBCollection collection = db.getCollection(ARCHIVEDRUNDOWNS);\r
 298                 DBCursor find = collection.find(new BasicDBObject(), new BasicDBObject(IOctopusAPI._ID, 0).append(IOctopusAPI.ID, 1));\r
 299                 if (find.hasNext())\r
 300                         result = ListUtils.cast(find.toArray());\r
 301                 return result;\r
 302         }\r
 303 \r
 304         private void setAndCheck(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,\r
 305                         int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine) throws Exception {\r
 306                 db = NoSQLUtils.getNoSQLDB();\r
 307                 if (db == null) {\r
 308                         logger.error(getMarker(), "Az NoSQL adatkezelő réteg nem elérhető.");\r
 309                         throw new NullPointerException("Internal error, missing NoSQL DB reference.");\r
 310                 }\r
 311 \r
 312                 if (jobEngine == null) {\r
 313                         logger.error(getMarker(), "Az folyamatkezelő réteg nem elérhető.");\r
 314                         throw new NullPointerException("Internal error, missing JobEngine reference.");\r
 315                 }\r
 316                 manager = jobEngine.getItemManager();\r
 317                 if (manager == null) {\r
 318                         logger.error(getMarker(), "Az adatbáziskezelő réteg nem elérhető.");\r
 319                         throw new NullPointerException("Internal error, missing ItemManager reference.");\r
 320                 }\r
 321                 String nexioHost = System.getProperty("nexio.host");\r
 322                 if (StringUtils.isBlank(nexioHost)) {\r
 323                         logger.error(getMarker(), "A 'nexio.host' rendszer paraméter nem található.");\r
 324                         throw new NullPointerException("System is not configured properly, 'jobengine.selenio.address' startup parameter missing.");\r
 325                 }\r
 326                 check(nexioPort, "nexioPort");\r
 327                 check(nexioUserName, "nexioUserName");\r
 328                 check(nexioPassword, "nexioPassword");\r
 329 \r
 330                 check(nexioKillDateDays, "nexioKillDateDays");\r
 331                 this.nexioKillDateDays = nexioKillDateDays;\r
 332                 check(nexioAgency, "nexioAgency");\r
 333                 this.nexioAgency = nexioAgency;\r
 334 \r
 335                 sourceUri = manager.createStoreUri(RemoteStoreProtocol.FTP, nexioHost);\r
 336                 sourceUri.setPortNumber(nexioPort);\r
 337                 sourceUri.setUserName(nexioUserName);\r
 338                 sourceUri.setPassword(nexioPassword);\r
 339 \r
 340                 check(archiveFtp, "archiveFtp");\r
 341                 check(archiveUserName, "archiveUserName");\r
 342                 check(archivePassword, "archivePassword");\r
 343 \r
 344                 targetUri = manager.createStoreUri(new URI(archiveFtp));\r
 345                 targetUri.setUserName(archiveUserName);\r
 346                 targetUri.setPassword(archivePassword);\r
 347 \r
 348         }\r
 349 \r
 350         private void transferFile(String fileName) throws Exception {\r
 351                 int reply = 0;\r
 352                 //              sourceFtp = ((FtpDirectoryLister) RemoteFileHandler.createLister(sourceUri)).connect();\r
 353                 //              targetFtp = ((FtpDirectoryLister) RemoteFileHandler.createLister(targetUri)).connect();\r
 354                 //              if (sourceUri != null)\r
 355                 //                      sourceUri.cleanUp();\r
 356                 //              if (targetUri != null)\r
 357                 //                      targetUri.cleanUp();\r
 358 \r
 359                 if (!targetFtp.enterRemotePassiveMode())\r
 360                         throw new Exception("!PASV");\r
 361 \r
 362                 reply = sourceFtp.port(InetAddress.getByName(targetFtp.getPassiveHost()), targetFtp.getPassivePort());\r
 363                 if (!FTPReply.isPositiveCompletion(reply))\r
 364                         throw new Exception("!PORT");\r
 365 \r
 366                 if (!sourceFtp.setFileType(FTP.BINARY_FILE_TYPE))\r
 367                         throw new Exception("!SOURCE TYPE");\r
 368 \r
 369                 sourceFtp.retr(fileName);\r
 370 \r
 371                 if (!targetFtp.setFileType(FTP.BINARY_FILE_TYPE))\r
 372                         throw new Exception("!TARGET TYPE");\r
 373 \r
 374                 targetFtp.stor(fileName);\r
 375 \r
 376                 while (true) {\r
 377                         reply = sourceFtp.stat();\r
 378                         if (!FTPReply.isPositiveCompletion(reply))\r
 379                                 throw new Exception("!STAT");\r
 380 \r
 381                         //String replyText = sourceFtp.getReplyString();\r
 382                         //logger.info("Status: {}", replyText);\r
 383                         if (reply == 226) {\r
 384                                 break;\r
 385                         }\r
 386                         Thread.sleep(100);\r
 387                 }\r
 388 \r
 389                 //              if (sourceUri != null)\r
 390                 //                      sourceUri.cleanUp();\r
 391                 //              if (targetUri != null)\r
 392                 //                      targetUri.cleanUp();\r
 393 \r
 394         }\r
 395 \r
 396         private void transferMetadata(String fileName, BasicDBObject metadata) throws Exception {\r
 397                 OutputStream outStream = null;\r
 398                 if (targetUri != null)\r
 399                         targetUri.cleanUp();\r
 400                 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();\r
 401                 try {\r
 402                         if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER)) {\r
 403                                 targetFtp.makeDirectory(EscortFiles.STATUSFOLDER);\r
 404                                 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER))\r
 405                                         throw new Exception("!STATUSFOLDER");\r
 406                         }\r
 407 \r
 408                         outStream = targetFtp.storeFileStream(fileName + JSON_EXT);\r
 409                         if (outStream == null) {\r
 410                                 throw new NullPointerException("Can not open: " + fileName + JSON_EXT + " Reply:" + targetFtp.getReplyString());\r
 411                         }\r
 412                         outStream.write(metadata.toString().getBytes(UTF_8));\r
 413                         outStream.flush();\r
 414                         targetFtp.changeToParentDirectory();\r
 415                 } catch (Exception e) {\r
 416                         logger.catching(e);\r
 417                         throw e;\r
 418                 } finally {\r
 419                         if (outStream != null)\r
 420                                 outStream.close();\r
 421                         //                      if (targetUri != null)\r
 422                         //                              targetUri.cleanUp();\r
 423                 }\r
 424         }\r
 425 \r
 426 }\r